Compare commits

..
Author SHA1 Message Date
longbingljwandGitHub 9f586c44d8 fix:env_file should read .env (#67)
* config adapt revert

* ci test

* fix mysql migration test

* fix

* fix

* lint fix

* fix ob config

* fix

* fix

* fix

* test over

* test

* fix

* fix

* fix style

* test over

* retain gin for pg

* gin for pg

* uuid defalut in versions

* ci test

* ci test

* fix

* fix

* fix

* fix

* pg josnb

* fix

* fix

* add seekdb

* test over

* test over

* fix env_fix
2025-11-17 10:52:59 +08:00
longbingljwandGitHub 5322f3bbd4 feat:add seekdb (#66)
* config adapt revert

* ci test

* fix mysql migration test

* fix

* fix

* lint fix

* fix ob config

* fix

* fix

* fix

* test over

* test

* fix

* fix

* fix style

* test over

* retain gin for pg

* gin for pg

* uuid defalut in versions

* ci test

* ci test

* fix

* fix

* fix

* fix

* pg josnb

* fix

* fix

* add seekdb

* test over

* test over
2025-11-16 18:39:35 +08:00
longbingljwandGitHub 6433ac8209 feat:json metadat filter adapt (#65)
* config adapt revert

* ci test

* fix mysql migration test

* fix

* fix

* lint fix

* fix ob config

* fix

* fix

* fix

* test over

* test

* fix

* fix

* fix style

* test over

* retain gin for pg

* gin for pg

* uuid defalut in versions

* ci test

* ci test

* fix

* fix

* fix

* fix

* pg josnb

* fix
2025-11-15 22:29:59 +08:00
longbingljwandGitHub 84935b9169 revert:public schema (#64)
* config adapt revert

* ci test

* fix mysql migration test

* fix

* fix

* lint fix

* fix ob config

* fix

* fix

* fix

* test over
2025-11-15 04:48:09 +08:00
longbingljwandGitHub eceaea68b1 fix:ci (#63)
* fix ci

* fix ci

* fix
2025-11-14 14:44:13 +08:00
longbingljwandGitHub 26cfccb84b fix (#62) 2025-11-14 11:20:05 +08:00
longbingljwandGitHub 3042b69e77 fix:style and ci (#60)
* mysql adaptation

* fix ci

* fix
2025-11-13 22:33:05 +08:00
longbingljwandGitHub 49d5637b3c mysql adaptation (#59) 2025-11-13 21:27:41 +08:00
231 changed files with 2184 additions and 8877 deletions
-5
View File
@@ -28,11 +28,6 @@ jobs:
# Format code
uv run ruff format ..
- name: count migration progress
run: |
cd api
./cnt_base.sh
- name: ast-grep
run: |
uvx --from ast-grep-cli sg --pattern 'db.session.query($WHATEVER).filter($HERE)' --rewrite 'db.session.query($WHATEVER).where($HERE)' -l py --update-all
+9 -9
View File
@@ -51,13 +51,13 @@ jobs:
- name: Expose Service Ports
run: sh .github/workflows/expose_service_ports.sh
# - name: Set up Vector Store (TiDB)
# uses: hoverkraft-tech/compose-action@v2.0.2
# with:
# compose-file: docker/tidb/docker-compose.yaml
# services: |
# tidb
# tiflash
- name: Set up Vector Store (TiDB)
uses: hoverkraft-tech/compose-action@v2.0.2
with:
compose-file: docker/tidb/docker-compose.yaml
services: |
tidb
tiflash
- name: Set up Vector Stores (Weaviate, Qdrant, PGVector, Milvus, PgVecto-RS, Chroma, MyScale, ElasticSearch, Couchbase, OceanBase)
uses: hoverkraft-tech/compose-action@v2.0.2
@@ -83,8 +83,8 @@ jobs:
ls -lah .
cp api/tests/integration_tests/.env.example api/tests/integration_tests/.env
# - name: Check VDB Ready (TiDB)
# run: uv run --project api python api/tests/integration_tests/vdb/tidb_vector/check_tiflash_ready.py
- name: Check VDB Ready (TiDB)
run: uv run --project api python api/tests/integration_tests/vdb/tidb_vector/check_tiflash_ready.py
- name: Test Vector Stores
run: uv run --project api bash dev/pytest/pytest_vdb.sh
+1 -7
View File
@@ -70,11 +70,6 @@ type-check:
@uv run --directory api --dev basedpyright
@echo "✅ Type check complete"
test:
@echo "🧪 Running backend unit tests..."
@uv run --project api --dev dev/pytest/pytest_unit_tests.sh
@echo "✅ Tests complete"
# Build Docker images
build-web:
@echo "Building web Docker image: $(WEB_IMAGE):$(VERSION)..."
@@ -124,7 +119,6 @@ help:
@echo " make check - Check code with ruff"
@echo " make lint - Format and fix code with ruff"
@echo " make type-check - Run type checking with basedpyright"
@echo " make test - Run backend unit tests"
@echo ""
@echo "Docker Build Targets:"
@echo " make build-web - Build web Docker image"
@@ -134,4 +128,4 @@ help:
@echo " make build-push-all - Build and push all Docker images"
# Phony targets
.PHONY: build-web build-api push-web push-api build-all push-all build-push-all dev-setup prepare-docker prepare-web prepare-api dev-clean help format check lint type-check test
.PHONY: build-web build-api push-web push-api build-all push-all build-push-all dev-setup prepare-docker prepare-web prepare-api dev-clean help format check lint type-check
+2 -1
View File
@@ -162,7 +162,8 @@ SUPABASE_URL=your-server-url
# CORS configuration
WEB_API_CORS_ALLOW_ORIGINS=http://localhost:3000,*
CONSOLE_CORS_ALLOW_ORIGINS=http://localhost:3000,*
# When the frontend and backend run on different subdomains, set COOKIE_DOMAIN to the sites top-level domain (e.g., `example.com`). Leading dots are optional.
# Set COOKIE_DOMAIN when the console frontend and API are on different subdomains.
# Provide the registrable domain (e.g. example.com); leading dots are optional.
COOKIE_DOMAIN=
# Vector database configuration
-4
View File
@@ -26,10 +26,6 @@
cp .env.example .env
```
> [!IMPORTANT]
>
> When the frontend and backend run on different subdomains, set COOKIE_DOMAIN to the sites top-level domain (e.g., `example.com`). The frontend and backend must be under the same top-level domain in order to share authentication cookies.
1. Generate a `SECRET_KEY` in the `.env` file.
bash for Linux
-7
View File
@@ -1,7 +0,0 @@
#!/bin/bash
set -euxo pipefail
for pattern in "Base" "TypeBase"; do
printf "%s " "$pattern"
grep "($pattern):" -r --include='*.py' --exclude-dir=".venv" --exclude-dir="tests" . | wc -l
done
+5 -1
View File
@@ -77,6 +77,10 @@ class AppExecutionConfig(BaseSettings):
description="Maximum number of concurrent active requests per app (0 for unlimited)",
default=0,
)
APP_DAILY_RATE_LIMIT: NonNegativeInt = Field(
description="Maximum number of requests per app per day",
default=5000,
)
class CodeExecutionSandboxConfig(BaseSettings):
@@ -1082,7 +1086,7 @@ class CeleryScheduleTasksConfig(BaseSettings):
)
TRIGGER_PROVIDER_CREDENTIAL_THRESHOLD_SECONDS: int = Field(
description="Proactive credential refresh threshold in seconds",
default=60 * 60,
default=180,
)
TRIGGER_PROVIDER_SUBSCRIPTION_THRESHOLD_SECONDS: int = Field(
description="Proactive subscription refresh threshold in seconds",
+3 -1
View File
@@ -250,8 +250,10 @@ class AppApi(Resource):
args = parser.parse_args()
app_service = AppService()
# Construct ArgsDict from parsed arguments
from services.app_service import AppService as AppServiceType
args_dict: AppService.ArgsDict = {
args_dict: AppServiceType.ArgsDict = {
"name": args["name"],
"description": args.get("description", ""),
"icon_type": args.get("icon_type", ""),
@@ -162,7 +162,6 @@ class DatasetDocumentListApi(Resource):
"keyword": "Search keyword",
"sort": "Sort order (default: -created_at)",
"fetch": "Fetch full details (default: false)",
"status": "Filter documents by display status",
}
)
@api.response(200, "Documents retrieved successfully")
@@ -176,7 +175,6 @@ class DatasetDocumentListApi(Resource):
limit = request.args.get("limit", default=20, type=int)
search = request.args.get("keyword", default=None, type=str)
sort = request.args.get("sort", default="-created_at", type=str)
status = request.args.get("status", default=None, type=str)
# "yes", "true", "t", "y", "1" convert to True, while others convert to False.
try:
fetch_val = request.args.get("fetch", default="false")
@@ -205,9 +203,6 @@ class DatasetDocumentListApi(Resource):
query = select(Document).filter_by(dataset_id=str(dataset_id), tenant_id=current_tenant_id)
if status:
query = DocumentService.apply_display_status_filter(query, status)
if search:
search = f"%{search}%"
query = query.where(Document.name.like(search))
@@ -1086,13 +1086,7 @@ class ToolMCPAuthApi(Resource):
return {"result": "success"}
except MCPAuthError as e:
try:
# Pass the extracted OAuth metadata hints to auth()
auth_result = auth(
provider_entity,
args.get("authorization_code"),
resource_metadata_url=e.resource_metadata_url,
scope_hint=e.scope_hint,
)
auth_result = auth(provider_entity, args.get("authorization_code"))
with Session(db.engine) as session, session.begin():
service = MCPToolManageService(session=session)
response = service.execute_auth_actions(auth_result)
@@ -1102,7 +1096,7 @@ class ToolMCPAuthApi(Resource):
service = MCPToolManageService(session=session)
service.clear_provider_credentials(provider_id=provider_id, tenant_id=tenant_id)
raise ValueError(f"Failed to refresh token, please try to authorize again: {e}") from e
except (MCPError, ValueError) as e:
except MCPError as e:
with Session(db.engine) as session, session.begin():
service = MCPToolManageService(session=session)
service.clear_provider_credentials(provider_id=provider_id, tenant_id=tenant_id)
@@ -128,7 +128,7 @@ class TenantApi(Resource):
@login_required
@account_initialization_required
@marshal_with(tenant_fields)
def post(self):
def get(self):
if request.path == "/info":
logger.warning("Deprecated URL /info was used.")
@@ -456,16 +456,12 @@ class DocumentListApi(DatasetApiResource):
page = request.args.get("page", default=1, type=int)
limit = request.args.get("limit", default=20, type=int)
search = request.args.get("keyword", default=None, type=str)
status = request.args.get("status", default=None, type=str)
dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
if not dataset:
raise NotFound("Dataset not found.")
query = select(Document).filter_by(dataset_id=str(dataset_id), tenant_id=tenant_id)
if status:
query = DocumentService.apply_display_status_filter(query, status)
if search:
search = f"%{search}%"
query = query.where(Document.name.like(search))
+1 -2
View File
@@ -145,8 +145,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
**extract_external_trace_id_from_args(args),
}
workflow_run_id = str(uuid.uuid4())
# FIXME (Yeuoly): we need to remove the SKIP_PREPARE_USER_INPUTS_KEY from the args
# trigger shouldn't prepare user inputs
# for trigger debug run, not prepare user inputs
if self._should_prepare_user_inputs(args):
inputs = self._prepare_user_inputs(
user_inputs=inputs,
@@ -1,10 +1,14 @@
from typing import Any
from typing import TYPE_CHECKING, Any, Optional
from pydantic import BaseModel, Field
# Import InvokeFrom locally to avoid circular import
from core.app.entities.app_invoke_entities import InvokeFrom
from core.datasource.entities.datasource_entities import DatasourceInvokeFrom
if TYPE_CHECKING:
from core.app.entities.app_invoke_entities import InvokeFrom
class DatasourceRuntime(BaseModel):
"""
@@ -13,7 +17,7 @@ class DatasourceRuntime(BaseModel):
tenant_id: str
datasource_id: str | None = None
invoke_from: InvokeFrom | None = None
invoke_from: Optional["InvokeFrom"] = None
datasource_invoke_from: DatasourceInvokeFrom | None = None
credentials: dict[str, Any] = Field(default_factory=dict)
runtime_parameters: dict[str, Any] = Field(default_factory=dict)
+49 -213
View File
@@ -6,8 +6,7 @@ import secrets
import urllib.parse
from urllib.parse import urljoin, urlparse
import httpx
from httpx import RequestError
from httpx import ConnectError, HTTPStatusError, RequestError
from pydantic import ValidationError
from core.entities.mcp_provider import MCPProviderEntity, MCPSupportGrantType
@@ -21,7 +20,6 @@ from core.mcp.types import (
OAuthClientMetadata,
OAuthMetadata,
OAuthTokens,
ProtectedResourceMetadata,
)
from extensions.ext_redis import redis_client
@@ -41,131 +39,6 @@ def generate_pkce_challenge() -> tuple[str, str]:
return code_verifier, code_challenge
def build_protected_resource_metadata_discovery_urls(
www_auth_resource_metadata_url: str | None, server_url: str
) -> list[str]:
"""
Build a list of URLs to try for Protected Resource Metadata discovery.
Per SEP-985, supports fallback when discovery fails at one URL.
"""
urls = []
# First priority: URL from WWW-Authenticate header
if www_auth_resource_metadata_url:
urls.append(www_auth_resource_metadata_url)
# Fallback: construct from server URL
parsed = urlparse(server_url)
base_url = f"{parsed.scheme}://{parsed.netloc}"
fallback_url = urljoin(base_url, "/.well-known/oauth-protected-resource")
if fallback_url not in urls:
urls.append(fallback_url)
return urls
def build_oauth_authorization_server_metadata_discovery_urls(auth_server_url: str | None, server_url: str) -> list[str]:
"""
Build a list of URLs to try for OAuth Authorization Server Metadata discovery.
Supports both OAuth 2.0 (RFC 8414) and OpenID Connect discovery.
Per RFC 8414 section 3:
- If issuer has no path: https://example.com/.well-known/oauth-authorization-server
- If issuer has path: https://example.com/.well-known/oauth-authorization-server{path}
Example:
- issuer: https://example.com/oauth
- metadata: https://example.com/.well-known/oauth-authorization-server/oauth
"""
urls = []
base_url = auth_server_url or server_url
parsed = urlparse(base_url)
base = f"{parsed.scheme}://{parsed.netloc}"
path = parsed.path.rstrip("/") # Remove trailing slash
# Try OpenID Connect discovery first (more common)
urls.append(urljoin(base + "/", ".well-known/openid-configuration"))
# OAuth 2.0 Authorization Server Metadata (RFC 8414)
# Include the path component if present in the issuer URL
if path:
urls.append(urljoin(base, f".well-known/oauth-authorization-server{path}"))
else:
urls.append(urljoin(base, ".well-known/oauth-authorization-server"))
return urls
def discover_protected_resource_metadata(
prm_url: str | None, server_url: str, protocol_version: str | None = None
) -> ProtectedResourceMetadata | None:
"""Discover OAuth 2.0 Protected Resource Metadata (RFC 9470)."""
urls = build_protected_resource_metadata_discovery_urls(prm_url, server_url)
headers = {"MCP-Protocol-Version": protocol_version or LATEST_PROTOCOL_VERSION, "User-Agent": "Dify"}
for url in urls:
try:
response = ssrf_proxy.get(url, headers=headers)
if response.status_code == 200:
return ProtectedResourceMetadata.model_validate(response.json())
elif response.status_code == 404:
continue # Try next URL
except (RequestError, ValidationError):
continue # Try next URL
return None
def discover_oauth_authorization_server_metadata(
auth_server_url: str | None, server_url: str, protocol_version: str | None = None
) -> OAuthMetadata | None:
"""Discover OAuth 2.0 Authorization Server Metadata (RFC 8414)."""
urls = build_oauth_authorization_server_metadata_discovery_urls(auth_server_url, server_url)
headers = {"MCP-Protocol-Version": protocol_version or LATEST_PROTOCOL_VERSION, "User-Agent": "Dify"}
for url in urls:
try:
response = ssrf_proxy.get(url, headers=headers)
if response.status_code == 200:
return OAuthMetadata.model_validate(response.json())
elif response.status_code == 404:
continue # Try next URL
except (RequestError, ValidationError):
continue # Try next URL
return None
def get_effective_scope(
scope_from_www_auth: str | None,
prm: ProtectedResourceMetadata | None,
asm: OAuthMetadata | None,
client_scope: str | None,
) -> str | None:
"""
Determine effective scope using priority-based selection strategy.
Priority order:
1. WWW-Authenticate header scope (server explicit requirement)
2. Protected Resource Metadata scopes
3. OAuth Authorization Server Metadata scopes
4. Client configured scope
"""
if scope_from_www_auth:
return scope_from_www_auth
if prm and prm.scopes_supported:
return " ".join(prm.scopes_supported)
if asm and asm.scopes_supported:
return " ".join(asm.scopes_supported)
return client_scope
def _create_secure_redis_state(state_data: OAuthCallbackState) -> str:
"""Create a secure state parameter by storing state data in Redis and returning a random state key."""
# Generate a secure random state key
@@ -248,36 +121,42 @@ def check_support_resource_discovery(server_url: str) -> tuple[bool, str]:
return False, ""
def discover_oauth_metadata(
server_url: str,
resource_metadata_url: str | None = None,
scope_hint: str | None = None,
protocol_version: str | None = None,
) -> tuple[OAuthMetadata | None, ProtectedResourceMetadata | None, str | None]:
"""
Discover OAuth metadata using RFC 8414/9470 standards.
def discover_oauth_metadata(server_url: str, protocol_version: str | None = None) -> OAuthMetadata | None:
"""Looks up RFC 8414 OAuth 2.0 Authorization Server Metadata."""
# First check if the server supports OAuth 2.0 Resource Discovery
support_resource_discovery, oauth_discovery_url = check_support_resource_discovery(server_url)
if support_resource_discovery:
# The oauth_discovery_url is the authorization server base URL
# Try OpenID Connect discovery first (more common), then OAuth 2.0
urls_to_try = [
urljoin(oauth_discovery_url + "/", ".well-known/oauth-authorization-server"),
urljoin(oauth_discovery_url + "/", ".well-known/openid-configuration"),
]
else:
urls_to_try = [urljoin(server_url, "/.well-known/oauth-authorization-server")]
Args:
server_url: The MCP server URL
resource_metadata_url: Protected Resource Metadata URL from WWW-Authenticate header
scope_hint: Scope hint from WWW-Authenticate header
protocol_version: MCP protocol version
headers = {"MCP-Protocol-Version": protocol_version or LATEST_PROTOCOL_VERSION}
Returns:
(oauth_metadata, protected_resource_metadata, scope_hint)
"""
# Discover Protected Resource Metadata
prm = discover_protected_resource_metadata(resource_metadata_url, server_url, protocol_version)
for url in urls_to_try:
try:
response = ssrf_proxy.get(url, headers=headers)
if response.status_code == 404:
continue
if not response.is_success:
response.raise_for_status()
return OAuthMetadata.model_validate(response.json())
except (RequestError, HTTPStatusError) as e:
if isinstance(e, ConnectError):
response = ssrf_proxy.get(url)
if response.status_code == 404:
continue # Try next URL
if not response.is_success:
raise ValueError(f"HTTP {response.status_code} trying to load well-known OAuth metadata")
return OAuthMetadata.model_validate(response.json())
# For other errors, try next URL
continue
# Get authorization server URL from PRM or use server URL
auth_server_url = None
if prm and prm.authorization_servers:
auth_server_url = prm.authorization_servers[0]
# Discover OAuth Authorization Server Metadata
asm = discover_oauth_authorization_server_metadata(auth_server_url, server_url, protocol_version)
return asm, prm, scope_hint
return None # No metadata found
def start_authorization(
@@ -287,7 +166,6 @@ def start_authorization(
redirect_url: str,
provider_id: str,
tenant_id: str,
scope: str | None = None,
) -> tuple[str, str]:
"""Begins the authorization flow with secure Redis state storage."""
response_type = "code"
@@ -297,6 +175,13 @@ def start_authorization(
authorization_url = metadata.authorization_endpoint
if response_type not in metadata.response_types_supported:
raise ValueError(f"Incompatible auth server: does not support response type {response_type}")
if (
not metadata.code_challenge_methods_supported
or code_challenge_method not in metadata.code_challenge_methods_supported
):
raise ValueError(
f"Incompatible auth server: does not support code challenge method {code_challenge_method}"
)
else:
authorization_url = urljoin(server_url, "/authorize")
@@ -325,49 +210,10 @@ def start_authorization(
"state": state_key,
}
# Add scope if provided
if scope:
params["scope"] = scope
authorization_url = f"{authorization_url}?{urllib.parse.urlencode(params)}"
return authorization_url, code_verifier
def _parse_token_response(response: httpx.Response) -> OAuthTokens:
"""
Parse OAuth token response supporting both JSON and form-urlencoded formats.
Per RFC 6749 Section 5.1, the standard format is JSON.
However, some legacy OAuth providers (e.g., early GitHub OAuth Apps) return
application/x-www-form-urlencoded format for backwards compatibility.
Args:
response: The HTTP response from token endpoint
Returns:
Parsed OAuth tokens
Raises:
ValueError: If response cannot be parsed
"""
content_type = response.headers.get("content-type", "").lower()
if "application/json" in content_type:
# Standard OAuth 2.0 JSON response (RFC 6749)
return OAuthTokens.model_validate(response.json())
elif "application/x-www-form-urlencoded" in content_type:
# Legacy form-urlencoded response (non-standard but used by some providers)
token_data = dict(urllib.parse.parse_qsl(response.text))
return OAuthTokens.model_validate(token_data)
else:
# No content-type or unknown - try JSON first, fallback to form-urlencoded
try:
return OAuthTokens.model_validate(response.json())
except (ValidationError, json.JSONDecodeError):
token_data = dict(urllib.parse.parse_qsl(response.text))
return OAuthTokens.model_validate(token_data)
def exchange_authorization(
server_url: str,
metadata: OAuthMetadata | None,
@@ -400,7 +246,7 @@ def exchange_authorization(
response = ssrf_proxy.post(token_url, data=params)
if not response.is_success:
raise ValueError(f"Token exchange failed: HTTP {response.status_code}")
return _parse_token_response(response)
return OAuthTokens.model_validate(response.json())
def refresh_authorization(
@@ -433,7 +279,7 @@ def refresh_authorization(
raise MCPRefreshTokenError(e) from e
if not response.is_success:
raise MCPRefreshTokenError(response.text)
return _parse_token_response(response)
return OAuthTokens.model_validate(response.json())
def client_credentials_flow(
@@ -476,7 +322,7 @@ def client_credentials_flow(
f"Client credentials token request failed: HTTP {response.status_code}, Response: {response.text}"
)
return _parse_token_response(response)
return OAuthTokens.model_validate(response.json())
def register_client(
@@ -506,8 +352,6 @@ def auth(
provider: MCPProviderEntity,
authorization_code: str | None = None,
state_param: str | None = None,
resource_metadata_url: str | None = None,
scope_hint: str | None = None,
) -> AuthResult:
"""
Orchestrates the full auth flow with a server using secure Redis state storage.
@@ -519,26 +363,18 @@ def auth(
provider: The MCP provider entity
authorization_code: Optional authorization code from OAuth callback
state_param: Optional state parameter from OAuth callback
resource_metadata_url: Optional Protected Resource Metadata URL from WWW-Authenticate
scope_hint: Optional scope hint from WWW-Authenticate header
Returns:
AuthResult containing actions to be performed and response data
"""
actions: list[AuthAction] = []
server_url = provider.decrypt_server_url()
# Discover OAuth metadata using RFC 8414/9470 standards
server_metadata, prm, scope_from_www_auth = discover_oauth_metadata(
server_url, resource_metadata_url, scope_hint, LATEST_PROTOCOL_VERSION
)
server_metadata = discover_oauth_metadata(server_url)
client_metadata = provider.client_metadata
provider_id = provider.id
tenant_id = provider.tenant_id
client_information = provider.retrieve_client_information()
redirect_url = provider.redirect_url
credentials = provider.decrypt_credentials()
# Determine grant type based on server metadata
if not server_metadata:
@@ -556,8 +392,8 @@ def auth(
else:
effective_grant_type = MCPSupportGrantType.CLIENT_CREDENTIALS.value
# Determine effective scope using priority-based strategy
effective_scope = get_effective_scope(scope_from_www_auth, prm, server_metadata, credentials.get("scope"))
# Get stored credentials
credentials = provider.decrypt_credentials()
if not client_information:
if authorization_code is not None:
@@ -589,11 +425,12 @@ def auth(
if effective_grant_type == MCPSupportGrantType.CLIENT_CREDENTIALS.value:
# Direct token request without user interaction
try:
scope = credentials.get("scope")
tokens = client_credentials_flow(
server_url,
server_metadata,
client_information,
effective_scope,
scope,
)
# Return action to save tokens and grant type
@@ -689,7 +526,6 @@ def auth(
redirect_url,
provider_id,
tenant_id,
effective_scope,
)
# Return action to save code verifier
+1 -7
View File
@@ -90,13 +90,7 @@ class MCPClientWithAuthRetry(MCPClient):
mcp_service = MCPToolManageService(session=session)
# Perform authentication using the service's auth method
# Extract OAuth metadata hints from the error
mcp_service.auth_with_actions(
self.provider_entity,
self.authorization_code,
resource_metadata_url=error.resource_metadata_url,
scope_hint=error.scope_hint,
)
mcp_service.auth_with_actions(self.provider_entity, self.authorization_code)
# Retrieve new tokens
self.provider_entity = mcp_service.get_provider_entity(
+1 -1
View File
@@ -290,7 +290,7 @@ def sse_client(
except httpx.HTTPStatusError as exc:
if exc.response.status_code == 401:
raise MCPAuthError(response=exc.response)
raise MCPAuthError()
raise MCPConnectionError()
except Exception:
logger.exception("Error connecting to SSE endpoint")
+1 -50
View File
@@ -1,10 +1,3 @@
import re
from typing import TYPE_CHECKING
if TYPE_CHECKING:
import httpx
class MCPError(Exception):
pass
@@ -14,49 +7,7 @@ class MCPConnectionError(MCPError):
class MCPAuthError(MCPConnectionError):
def __init__(
self,
message: str | None = None,
response: "httpx.Response | None" = None,
www_authenticate_header: str | None = None,
):
"""
MCP Authentication Error.
Args:
message: Error message
response: HTTP response object (will extract WWW-Authenticate header if provided)
www_authenticate_header: Pre-extracted WWW-Authenticate header value
"""
super().__init__(message or "Authentication failed")
# Extract OAuth metadata hints from WWW-Authenticate header
if response is not None:
www_authenticate_header = response.headers.get("WWW-Authenticate")
self.resource_metadata_url: str | None = None
self.scope_hint: str | None = None
if www_authenticate_header:
self.resource_metadata_url = self._extract_field(www_authenticate_header, "resource_metadata")
self.scope_hint = self._extract_field(www_authenticate_header, "scope")
@staticmethod
def _extract_field(www_auth: str, field_name: str) -> str | None:
"""Extract a specific field from the WWW-Authenticate header."""
# Pattern to match field="value" or field=value
pattern = rf'{field_name}="([^"]*)"'
match = re.search(pattern, www_auth)
if match:
return match.group(1)
# Try without quotes
pattern = rf"{field_name}=([^\s,]+)"
match = re.search(pattern, www_auth)
if match:
return match.group(1)
return None
pass
class MCPRefreshTokenError(MCPError):
+11 -21
View File
@@ -149,7 +149,7 @@ class BaseSession(
messages when entered.
"""
_response_streams: dict[RequestId, queue.Queue[JSONRPCResponse | JSONRPCError | HTTPStatusError]]
_response_streams: dict[RequestId, queue.Queue[JSONRPCResponse | JSONRPCError]]
_request_id: int
_in_flight: dict[RequestId, RequestResponder[ReceiveRequestT, SendResultT]]
_receive_request_type: type[ReceiveRequestT]
@@ -230,7 +230,7 @@ class BaseSession(
request_id = self._request_id
self._request_id = request_id + 1
response_queue: queue.Queue[JSONRPCResponse | JSONRPCError | HTTPStatusError] = queue.Queue()
response_queue: queue.Queue[JSONRPCResponse | JSONRPCError] = queue.Queue()
self._response_streams[request_id] = response_queue
try:
@@ -261,17 +261,11 @@ class BaseSession(
message="No response received",
)
)
elif isinstance(response_or_error, HTTPStatusError):
# HTTPStatusError from streamable_client with preserved response object
if response_or_error.response.status_code == 401:
raise MCPAuthError(response=response_or_error.response)
else:
raise MCPConnectionError(
ErrorData(code=response_or_error.response.status_code, message=str(response_or_error))
)
elif isinstance(response_or_error, JSONRPCError):
if response_or_error.error.code == 401:
raise MCPAuthError(message=response_or_error.error.message)
raise MCPAuthError(
ErrorData(code=response_or_error.error.code, message=response_or_error.error.message)
)
else:
raise MCPConnectionError(
ErrorData(code=response_or_error.error.code, message=response_or_error.error.message)
@@ -333,17 +327,13 @@ class BaseSession(
if isinstance(message, HTTPStatusError):
response_queue = self._response_streams.get(self._request_id - 1)
if response_queue is not None:
# For 401 errors, pass the HTTPStatusError directly to preserve response object
if message.response.status_code == 401:
response_queue.put(message)
else:
response_queue.put(
JSONRPCError(
jsonrpc="2.0",
id=self._request_id - 1,
error=ErrorData(code=message.response.status_code, message=message.args[0]),
)
response_queue.put(
JSONRPCError(
jsonrpc="2.0",
id=self._request_id - 1,
error=ErrorData(code=message.response.status_code, message=message.args[0]),
)
)
else:
self._handle_incoming(RuntimeError(f"Received response with an unknown request ID: {message}"))
elif isinstance(message, Exception):
+1 -11
View File
@@ -23,7 +23,7 @@ for reference.
not separate types in the schema.
"""
# Client support both version, not support 2025-06-18 yet.
LATEST_PROTOCOL_VERSION = "2025-06-18"
LATEST_PROTOCOL_VERSION = "2025-03-26"
# Server support 2024-11-05 to allow claude to use.
SERVER_LATEST_PROTOCOL_VERSION = "2024-11-05"
DEFAULT_NEGOTIATED_VERSION = "2025-03-26"
@@ -1330,13 +1330,3 @@ class OAuthMetadata(BaseModel):
response_types_supported: list[str]
grant_types_supported: list[str] | None = None
code_challenge_methods_supported: list[str] | None = None
scopes_supported: list[str] | None = None
class ProtectedResourceMetadata(BaseModel):
"""OAuth 2.0 Protected Resource Metadata (RFC 9470)."""
resource: str | None = None
authorization_servers: list[str]
scopes_supported: list[str] | None = None
bearer_methods_supported: list[str] | None = None
+1
View File
@@ -63,6 +63,7 @@ from services.tools.tools_transform_service import ToolTransformService
if TYPE_CHECKING:
from core.workflow.nodes.tool.entities import ToolEntity
from core.workflow.runtime import VariablePool
logger = logging.getLogger(__name__)
@@ -192,6 +192,7 @@ class GraphEngine:
self._dispatcher = Dispatcher(
event_queue=self._event_queue,
event_handler=self._event_handler_registry,
event_collector=self._event_manager,
execution_coordinator=self._execution_coordinator,
event_emitter=self._event_manager,
)
@@ -43,6 +43,7 @@ class Dispatcher:
self,
event_queue: queue.Queue[GraphNodeEventBase],
event_handler: "EventHandler",
event_collector: EventManager,
execution_coordinator: ExecutionCoordinator,
event_emitter: EventManager | None = None,
) -> None:
@@ -52,11 +53,13 @@ class Dispatcher:
Args:
event_queue: Queue of events from workers
event_handler: Event handler registry for processing events
event_collector: Event manager for collecting unhandled events
execution_coordinator: Coordinator for execution flow
event_emitter: Optional event manager to signal completion
"""
self._event_queue = event_queue
self._event_handler = event_handler
self._event_collector = event_collector
self._execution_coordinator = execution_coordinator
self._event_emitter = event_emitter
@@ -83,31 +86,37 @@ class Dispatcher:
def _dispatcher_loop(self) -> None:
"""Main dispatcher loop."""
try:
self._process_commands()
while not self._stop_event.is_set():
if (
self._execution_coordinator.aborted
or self._execution_coordinator.paused
or self._execution_coordinator.execution_complete
):
break
commands_checked = False
should_check_commands = False
should_break = False
self._execution_coordinator.check_scaling()
try:
event = self._event_queue.get(timeout=0.1)
self._event_handler.dispatch(event)
self._event_queue.task_done()
self._process_commands(event)
except queue.Empty:
time.sleep(0.1)
if self._execution_coordinator.is_execution_complete():
should_check_commands = True
should_break = True
else:
# Check for scaling
self._execution_coordinator.check_scaling()
self._process_commands()
while True:
try:
event = self._event_queue.get(block=False)
self._event_handler.dispatch(event)
self._event_queue.task_done()
except queue.Empty:
# Process events
try:
event = self._event_queue.get(timeout=0.1)
# Route to the event handler
self._event_handler.dispatch(event)
should_check_commands = self._should_check_commands(event)
self._event_queue.task_done()
except queue.Empty:
# Process commands even when no new events arrive so abort requests are not missed
should_check_commands = True
time.sleep(0.1)
if should_check_commands and not commands_checked:
self._execution_coordinator.check_commands()
commands_checked = True
if should_break:
if not commands_checked:
self._execution_coordinator.check_commands()
break
except Exception as e:
@@ -120,6 +129,6 @@ class Dispatcher:
if self._event_emitter:
self._event_emitter.mark_complete()
def _process_commands(self, event: GraphNodeEventBase | None = None):
if event is None or isinstance(event, self._COMMAND_TRIGGER_EVENTS):
self._execution_coordinator.process_commands()
def _should_check_commands(self, event: GraphNodeEventBase) -> bool:
"""Return True if the event represents a node completion."""
return isinstance(event, self._COMMAND_TRIGGER_EVENTS)
@@ -40,7 +40,7 @@ class ExecutionCoordinator:
self._command_processor = command_processor
self._worker_pool = worker_pool
def process_commands(self) -> None:
def check_commands(self) -> None:
"""Process any pending commands."""
self._command_processor.process_commands()
@@ -48,16 +48,24 @@ class ExecutionCoordinator:
"""Check and perform worker scaling if needed."""
self._worker_pool.check_and_scale()
@property
def execution_complete(self):
def is_execution_complete(self) -> bool:
"""
Check if execution is complete.
Returns:
True if execution is complete
"""
# Treat paused, aborted, or failed executions as terminal states
if self._graph_execution.is_paused:
return True
if self._graph_execution.aborted or self._graph_execution.has_error:
return True
return self._state_manager.is_execution_complete()
@property
def aborted(self):
return self._graph_execution.aborted or self._graph_execution.has_error
@property
def paused(self) -> bool:
def is_paused(self) -> bool:
"""Expose whether the underlying graph execution is paused."""
return self._graph_execution.is_paused
@@ -11,6 +11,7 @@ from sqlalchemy.orm import sessionmaker
from core.app.app_config.entities import DatasetRetrieveConfigEntity
from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
from core.callback_handler.index_tool_callback_handler import DatasetDocument
from core.entities.agent_entities import PlanningStrategy
from core.entities.model_entities import ModelStatus
from core.model_manager import ModelInstance, ModelManager
@@ -596,7 +597,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node):
if value is None and condition not in ("empty", "not empty"):
return filters
json_field = Document.doc_metadata[metadata_name].as_string()
json_field = DatasetDocument.doc_metadata[metadata_name].as_string()
match condition:
case "contains":
@@ -640,31 +641,31 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node):
if isinstance(value, str):
filters.append(json_field == value)
elif isinstance(value, (int, float)):
filters.append(Document.doc_metadata[metadata_name].as_float() == value)
filters.append(DatasetDocument.doc_metadata[metadata_name].as_float() == value)
case "is not" | "":
if isinstance(value, str):
filters.append(json_field != value)
elif isinstance(value, (int, float)):
filters.append(Document.doc_metadata[metadata_name].as_float() != value)
filters.append(DatasetDocument.doc_metadata[metadata_name].as_float() != value)
case "empty":
filters.append(Document.doc_metadata[metadata_name].is_(None))
filters.append(DatasetDocument.doc_metadata[metadata_name].is_(None))
case "not empty":
filters.append(Document.doc_metadata[metadata_name].isnot(None))
filters.append(DatasetDocument.doc_metadata[metadata_name].isnot(None))
case "before" | "<":
filters.append(Document.doc_metadata[metadata_name].as_float() < value)
filters.append(DatasetDocument.doc_metadata[metadata_name].as_float() < value)
case "after" | ">":
filters.append(Document.doc_metadata[metadata_name].as_float() > value)
filters.append(DatasetDocument.doc_metadata[metadata_name].as_float() > value)
case "" | "<=":
filters.append(Document.doc_metadata[metadata_name].as_float() <= value)
filters.append(DatasetDocument.doc_metadata[metadata_name].as_float() <= value)
case "" | ">=":
filters.append(Document.doc_metadata[metadata_name].as_float() >= value)
filters.append(DatasetDocument.doc_metadata[metadata_name].as_float() >= value)
case _:
pass
@@ -3,6 +3,7 @@ from __future__ import annotations
import importlib
import json
from collections.abc import Mapping, Sequence
from collections.abc import Mapping as TypingMapping
from copy import deepcopy
from dataclasses import dataclass
from typing import Any, Protocol
@@ -99,8 +100,8 @@ class ResponseStreamCoordinatorProtocol(Protocol):
class GraphProtocol(Protocol):
"""Structural interface required from graph instances attached to the runtime state."""
nodes: Mapping[str, object]
edges: Mapping[str, object]
nodes: TypingMapping[str, object]
edges: TypingMapping[str, object]
root_node: object
def get_outgoing_edges(self, node_id: str) -> Sequence[object]: ...
-209
View File
@@ -1,209 +0,0 @@
import logging
from dataclasses import dataclass
from enum import StrEnum, auto
logger = logging.getLogger(__name__)
@dataclass
class QuotaCharge:
"""
Result of a quota consumption operation.
Attributes:
success: Whether the quota charge succeeded
charge_id: UUID for refund, or None if failed/disabled
"""
success: bool
charge_id: str | None
_quota_type: "QuotaType"
def refund(self) -> None:
"""
Refund this quota charge.
Safe to call even if charge failed or was disabled.
This method guarantees no exceptions will be raised.
"""
if self.charge_id:
self._quota_type.refund(self.charge_id)
logger.info("Refunded quota for %s with charge_id: %s", self._quota_type.value, self.charge_id)
class QuotaType(StrEnum):
"""
Supported quota types for tenant feature usage.
Add additional types here whenever new billable features become available.
"""
# Trigger execution quota
TRIGGER = auto()
# Workflow execution quota
WORKFLOW = auto()
UNLIMITED = auto()
@property
def billing_key(self) -> str:
"""
Get the billing key for the feature.
"""
match self:
case QuotaType.TRIGGER:
return "trigger_event"
case QuotaType.WORKFLOW:
return "api_rate_limit"
case _:
raise ValueError(f"Invalid quota type: {self}")
def consume(self, tenant_id: str, amount: int = 1) -> QuotaCharge:
"""
Consume quota for the feature.
Args:
tenant_id: The tenant identifier
amount: Amount to consume (default: 1)
Returns:
QuotaCharge with success status and charge_id for refund
Raises:
QuotaExceededError: When quota is insufficient
"""
from configs import dify_config
from services.billing_service import BillingService
from services.errors.app import QuotaExceededError
if not dify_config.BILLING_ENABLED:
logger.debug("Billing disabled, allowing request for %s", tenant_id)
return QuotaCharge(success=True, charge_id=None, _quota_type=self)
logger.info("Consuming %d %s quota for tenant %s", amount, self.value, tenant_id)
if amount <= 0:
raise ValueError("Amount to consume must be greater than 0")
try:
response = BillingService.update_tenant_feature_plan_usage(tenant_id, self.billing_key, delta=amount)
if response.get("result") != "success":
logger.warning(
"Failed to consume quota for %s, feature %s details: %s",
tenant_id,
self.value,
response.get("detail"),
)
raise QuotaExceededError(feature=self.value, tenant_id=tenant_id, required=amount)
charge_id = response.get("history_id")
logger.debug(
"Successfully consumed %d %s quota for tenant %s, charge_id: %s",
amount,
self.value,
tenant_id,
charge_id,
)
return QuotaCharge(success=True, charge_id=charge_id, _quota_type=self)
except QuotaExceededError:
raise
except Exception:
# fail-safe: allow request on billing errors
logger.exception("Failed to consume quota for %s, feature %s", tenant_id, self.value)
return unlimited()
def check(self, tenant_id: str, amount: int = 1) -> bool:
"""
Check if tenant has sufficient quota without consuming.
Args:
tenant_id: The tenant identifier
amount: Amount to check (default: 1)
Returns:
True if quota is sufficient, False otherwise
"""
from configs import dify_config
if not dify_config.BILLING_ENABLED:
return True
if amount <= 0:
raise ValueError("Amount to check must be greater than 0")
try:
remaining = self.get_remaining(tenant_id)
return remaining >= amount if remaining != -1 else True
except Exception:
logger.exception("Failed to check quota for %s, feature %s", tenant_id, self.value)
# fail-safe: allow request on billing errors
return True
def refund(self, charge_id: str) -> None:
"""
Refund quota using charge_id from consume().
This method guarantees no exceptions will be raised.
All errors are logged but silently handled.
Args:
charge_id: The UUID returned from consume()
"""
try:
from configs import dify_config
from services.billing_service import BillingService
if not dify_config.BILLING_ENABLED:
return
if not charge_id:
logger.warning("Cannot refund: charge_id is empty")
return
logger.info("Refunding %s quota with charge_id: %s", self.value, charge_id)
response = BillingService.refund_tenant_feature_plan_usage(charge_id)
if response.get("result") == "success":
logger.debug("Successfully refunded %s quota, charge_id: %s", self.value, charge_id)
else:
logger.warning("Refund failed for charge_id: %s", charge_id)
except Exception:
# Catch ALL exceptions - refund must never fail
logger.exception("Failed to refund quota for charge_id: %s", charge_id)
# Don't raise - refund is best-effort and must be silent
def get_remaining(self, tenant_id: str) -> int:
"""
Get remaining quota for the tenant.
Args:
tenant_id: The tenant identifier
Returns:
Remaining quota amount
"""
from services.billing_service import BillingService
try:
usage_info = BillingService.get_tenant_feature_plan_usage(tenant_id, self.billing_key)
# Assuming the API returns a dict with 'remaining' or 'limit' and 'used'
if isinstance(usage_info, dict):
return usage_info.get("remaining", 0)
# If it returns a simple number, treat it as remaining
return int(usage_info) if usage_info else 0
except Exception:
logger.exception("Failed to get remaining quota for %s, feature %s", tenant_id, self.value)
return -1
def unlimited() -> QuotaCharge:
"""
Return a quota charge for unlimited quota.
This is useful for features that are not subject to quota limits, such as the UNLIMITED quota type.
"""
return QuotaCharge(success=True, charge_id=None, _quota_type=QuotaType.UNLIMITED)
+1
View File
@@ -10,6 +10,7 @@ from redis import RedisError
from redis.cache import CacheConfig
from redis.cluster import ClusterNode, RedisCluster
from redis.connection import Connection, SSLConnection
from redis.lock import Lock
from redis.sentinel import Sentinel
from configs import dify_config
@@ -45,6 +45,7 @@ class ClickZettaVolumeConfig(BaseModel):
This method will first try to use CLICKZETTA_VOLUME_* environment variables,
then fall back to CLICKZETTA_* environment variables (for vector DB config).
"""
import os
# Helper function to get environment variable with fallback
def get_env_with_fallback(volume_key: str, fallback_key: str, default: str | None = None) -> str:
@@ -3,7 +3,7 @@ import io
import json
from collections.abc import Generator
from google.cloud import storage as google_cloud_storage # type: ignore
from google.cloud import storage as google_cloud_storage
from configs import dify_config
from extensions.storage.base_storage import BaseStorage
-78
View File
@@ -38,12 +38,6 @@ class EmailType(StrEnum):
EMAIL_REGISTER = auto()
EMAIL_REGISTER_WHEN_ACCOUNT_EXIST = auto()
RESET_PASSWORD_WHEN_ACCOUNT_NOT_EXIST_NO_REGISTER = auto()
TRIGGER_EVENTS_LIMIT_SANDBOX = auto()
TRIGGER_EVENTS_LIMIT_PROFESSIONAL = auto()
TRIGGER_EVENTS_USAGE_WARNING_SANDBOX = auto()
TRIGGER_EVENTS_USAGE_WARNING_PROFESSIONAL = auto()
API_RATE_LIMIT_LIMIT_SANDBOX = auto()
API_RATE_LIMIT_WARNING_SANDBOX = auto()
class EmailLanguage(StrEnum):
@@ -451,78 +445,6 @@ def create_default_email_config() -> EmailI18nConfig:
branded_template_path="clean_document_job_mail_template_zh-CN.html",
),
},
EmailType.TRIGGER_EVENTS_LIMIT_SANDBOX: {
EmailLanguage.EN_US: EmailTemplate(
subject="Youve reached your Sandbox Trigger Events limit",
template_path="trigger_events_limit_template_en-US.html",
branded_template_path="without-brand/trigger_events_limit_template_en-US.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="您的 Sandbox 触发事件额度已用尽",
template_path="trigger_events_limit_template_zh-CN.html",
branded_template_path="without-brand/trigger_events_limit_template_zh-CN.html",
),
},
EmailType.TRIGGER_EVENTS_LIMIT_PROFESSIONAL: {
EmailLanguage.EN_US: EmailTemplate(
subject="Youve reached your monthly Trigger Events limit",
template_path="trigger_events_limit_template_en-US.html",
branded_template_path="without-brand/trigger_events_limit_template_en-US.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="您的月度触发事件额度已用尽",
template_path="trigger_events_limit_template_zh-CN.html",
branded_template_path="without-brand/trigger_events_limit_template_zh-CN.html",
),
},
EmailType.TRIGGER_EVENTS_USAGE_WARNING_SANDBOX: {
EmailLanguage.EN_US: EmailTemplate(
subject="Youre nearing your Sandbox Trigger Events limit",
template_path="trigger_events_usage_warning_template_en-US.html",
branded_template_path="without-brand/trigger_events_usage_warning_template_en-US.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="您的 Sandbox 触发事件额度接近上限",
template_path="trigger_events_usage_warning_template_zh-CN.html",
branded_template_path="without-brand/trigger_events_usage_warning_template_zh-CN.html",
),
},
EmailType.TRIGGER_EVENTS_USAGE_WARNING_PROFESSIONAL: {
EmailLanguage.EN_US: EmailTemplate(
subject="Youre nearing your Monthly Trigger Events limit",
template_path="trigger_events_usage_warning_template_en-US.html",
branded_template_path="without-brand/trigger_events_usage_warning_template_en-US.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="您的月度触发事件额度接近上限",
template_path="trigger_events_usage_warning_template_zh-CN.html",
branded_template_path="without-brand/trigger_events_usage_warning_template_zh-CN.html",
),
},
EmailType.API_RATE_LIMIT_LIMIT_SANDBOX: {
EmailLanguage.EN_US: EmailTemplate(
subject="Youve reached your API Rate Limit",
template_path="api_rate_limit_limit_template_en-US.html",
branded_template_path="without-brand/api_rate_limit_limit_template_en-US.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="您的 API 速率额度已用尽",
template_path="api_rate_limit_limit_template_zh-CN.html",
branded_template_path="without-brand/api_rate_limit_limit_template_zh-CN.html",
),
},
EmailType.API_RATE_LIMIT_WARNING_SANDBOX: {
EmailLanguage.EN_US: EmailTemplate(
subject="Youre nearing your API Rate Limit",
template_path="api_rate_limit_warning_template_en-US.html",
branded_template_path="without-brand/api_rate_limit_warning_template_en-US.html",
),
EmailLanguage.ZH_HANS: EmailTemplate(
subject="您的 API 速率额度接近上限",
template_path="api_rate_limit_warning_template_zh-CN.html",
branded_template_path="without-brand/api_rate_limit_warning_template_zh-CN.html",
),
},
EmailType.EMAIL_REGISTER: {
EmailLanguage.EN_US: EmailTemplate(
subject="Register Your {application_title} Account",
+29 -41
View File
@@ -226,7 +226,7 @@ class Dataset(Base):
ExternalKnowledgeApis.id == external_knowledge_binding.external_knowledge_api_id
)
)
if external_knowledge_api is None or external_knowledge_api.settings is None:
if not external_knowledge_api:
return None
return {
"external_knowledge_id": external_knowledge_binding.external_knowledge_id,
@@ -944,19 +944,17 @@ class DatasetQuery(Base):
created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=sa.func.current_timestamp())
class DatasetKeywordTable(TypeBase):
class DatasetKeywordTable(Base):
__tablename__ = "dataset_keyword_tables"
__table_args__ = (
sa.PrimaryKeyConstraint("id", name="dataset_keyword_table_pkey"),
sa.Index("dataset_keyword_table_dataset_id_idx", "dataset_id"),
)
id: Mapped[str] = mapped_column(StringUUID, primary_key=True, default=lambda: str(uuid4()), init=False)
dataset_id: Mapped[str] = mapped_column(StringUUID, nullable=False, unique=True)
keyword_table: Mapped[str] = mapped_column(LongText, nullable=False)
data_source_type: Mapped[str] = mapped_column(
String(255), nullable=False, server_default=sa.text("'database'"), default="database"
)
id = mapped_column(StringUUID, primary_key=True, default=lambda: str(uuid4()))
dataset_id = mapped_column(StringUUID, nullable=False, unique=True)
keyword_table = mapped_column(LongText, nullable=False)
data_source_type = mapped_column(String(255), nullable=False, server_default=sa.text("'database'"))
@property
def keyword_table_dict(self) -> dict[str, set[Any]] | None:
@@ -1051,21 +1049,19 @@ class TidbAuthBinding(Base):
created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=func.current_timestamp())
class Whitelist(TypeBase):
class Whitelist(Base):
__tablename__ = "whitelists"
__table_args__ = (
sa.PrimaryKeyConstraint("id", name="whitelists_pkey"),
sa.Index("whitelists_tenant_idx", "tenant_id"),
)
id: Mapped[str] = mapped_column(StringUUID, primary_key=True, default=lambda: str(uuid4()), init=False)
tenant_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
id = mapped_column(StringUUID, primary_key=True, default=lambda: str(uuid4()))
tenant_id = mapped_column(StringUUID, nullable=True)
category: Mapped[str] = mapped_column(String(255), nullable=False)
created_at: Mapped[datetime] = mapped_column(
DateTime, nullable=False, server_default=func.current_timestamp(), init=False
)
created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=func.current_timestamp())
class DatasetPermission(TypeBase):
class DatasetPermission(Base):
__tablename__ = "dataset_permissions"
__table_args__ = (
sa.PrimaryKeyConstraint("id", name="dataset_permission_pkey"),
@@ -1074,19 +1070,15 @@ class DatasetPermission(TypeBase):
sa.Index("idx_dataset_permissions_tenant_id", "tenant_id"),
)
id: Mapped[str] = mapped_column(StringUUID, default=lambda: str(uuid4()), primary_key=True, init=False)
dataset_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
account_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
has_permission: Mapped[bool] = mapped_column(
sa.Boolean, nullable=False, server_default=sa.text("true"), default=True
)
created_at: Mapped[datetime] = mapped_column(
DateTime, nullable=False, server_default=func.current_timestamp(), init=False
)
id = mapped_column(StringUUID, default=lambda: str(uuid4()), primary_key=True)
dataset_id = mapped_column(StringUUID, nullable=False)
account_id = mapped_column(StringUUID, nullable=False)
tenant_id = mapped_column(StringUUID, nullable=False)
has_permission: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("true"))
created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=func.current_timestamp())
class ExternalKnowledgeApis(TypeBase):
class ExternalKnowledgeApis(Base):
__tablename__ = "external_knowledge_apis"
__table_args__ = (
sa.PrimaryKeyConstraint("id", name="external_knowledge_apis_pkey"),
@@ -1094,18 +1086,16 @@ class ExternalKnowledgeApis(TypeBase):
sa.Index("external_knowledge_apis_name_idx", "name"),
)
id: Mapped[str] = mapped_column(StringUUID, nullable=False, default=lambda: str(uuid4()), init=False)
id = mapped_column(StringUUID, nullable=False, default=lambda: str(uuid4()))
name: Mapped[str] = mapped_column(String(255), nullable=False)
description: Mapped[str] = mapped_column(String(255), nullable=False)
tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
settings: Mapped[str | None] = mapped_column(LongText, nullable=True)
created_by: Mapped[str] = mapped_column(StringUUID, nullable=False)
created_at: Mapped[datetime] = mapped_column(
DateTime, nullable=False, server_default=func.current_timestamp(), init=False
)
updated_by: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
tenant_id = mapped_column(StringUUID, nullable=False)
settings = mapped_column(LongText, nullable=True)
created_by = mapped_column(StringUUID, nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=func.current_timestamp())
updated_by = mapped_column(StringUUID, nullable=True)
updated_at: Mapped[datetime] = mapped_column(
DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp(), init=False
DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
)
def to_dict(self) -> dict[str, Any]:
@@ -1181,7 +1171,7 @@ class DatasetAutoDisableLog(Base):
created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=sa.func.current_timestamp())
class RateLimitLog(TypeBase):
class RateLimitLog(Base):
__tablename__ = "rate_limit_logs"
__table_args__ = (
sa.PrimaryKeyConstraint("id", name="rate_limit_log_pkey"),
@@ -1189,13 +1179,11 @@ class RateLimitLog(TypeBase):
sa.Index("rate_limit_log_operation_idx", "operation"),
)
id: Mapped[str] = mapped_column(StringUUID, default=lambda: str(uuid4()), init=False)
tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
id = mapped_column(StringUUID, default=lambda: str(uuid4()))
tenant_id = mapped_column(StringUUID, nullable=False)
subscription_plan: Mapped[str] = mapped_column(String(255), nullable=False)
operation: Mapped[str] = mapped_column(String(255), nullable=False)
created_at: Mapped[datetime] = mapped_column(
DateTime, nullable=False, server_default=func.current_timestamp(), init=False
)
created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=sa.func.current_timestamp())
class DatasetMetadata(Base):
-1
View File
@@ -64,7 +64,6 @@ class AppTriggerStatus(StrEnum):
ENABLED = "enabled"
DISABLED = "disabled"
UNAUTHORIZED = "unauthorized"
RATE_LIMITED = "rate_limited"
class AppTriggerType(StrEnum):
+11
View File
@@ -20,6 +20,9 @@ from .types import LongText, StringUUID
if TYPE_CHECKING:
from core.entities.mcp_provider import MCPProviderEntity
from core.tools.entities.common_entities import I18nObject
from core.tools.entities.tool_bundle import ApiToolBundle
from core.tools.entities.tool_entities import ApiProviderSchemaType, WorkflowToolParameterConfiguration
# system level tool oauth client params (client_id, client_secret, etc.)
@@ -160,10 +163,14 @@ class ApiToolProvider(TypeBase):
@property
def schema_type(self) -> "ApiProviderSchemaType":
from core.tools.entities.tool_entities import ApiProviderSchemaType
return ApiProviderSchemaType.value_of(self.schema_type_str)
@property
def tools(self) -> list["ApiToolBundle"]:
from core.tools.entities.tool_bundle import ApiToolBundle
return [ApiToolBundle.model_validate(tool) for tool in json.loads(self.tools_str)]
@property
@@ -256,6 +263,8 @@ class WorkflowToolProvider(TypeBase):
@property
def parameter_configurations(self) -> list["WorkflowToolParameterConfiguration"]:
from core.tools.entities.tool_entities import WorkflowToolParameterConfiguration
return [
WorkflowToolParameterConfiguration.model_validate(config)
for config in json.loads(self.parameter_configuration)
@@ -512,4 +521,6 @@ class DeprecatedPublishedAppTool(TypeBase):
@property
def description_i18n(self) -> "I18nObject":
from core.tools.entities.common_entities import I18nObject
return I18nObject.model_validate(json.loads(self.description))
+5 -7
View File
@@ -16,7 +16,7 @@ from core.trigger.entities.entities import Subscription
from core.trigger.utils.endpoint import generate_plugin_trigger_endpoint_url, generate_webhook_trigger_endpoint
from libs.datetime_utils import naive_utc_now
from libs.uuid_utils import uuidv7
from models.base import Base, TypeBase
from models.base import Base
from models.engine import db
from models.enums import AppTriggerStatus, AppTriggerType, CreatorUserRole, WorkflowTriggerStatus
from models.model import Account
@@ -401,7 +401,7 @@ class AppTrigger(Base):
)
class WorkflowSchedulePlan(TypeBase):
class WorkflowSchedulePlan(Base):
"""
Workflow Schedule Configuration
@@ -427,7 +427,7 @@ class WorkflowSchedulePlan(TypeBase):
sa.Index("workflow_schedule_plan_next_idx", "next_run_at"),
)
id: Mapped[str] = mapped_column(StringUUID, primary_key=True, default=lambda: str(uuidv7()), init=False)
id: Mapped[str] = mapped_column(StringUUID, default=lambda: str(uuidv7()))
app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
node_id: Mapped[str] = mapped_column(String(64), nullable=False)
tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
@@ -438,11 +438,9 @@ class WorkflowSchedulePlan(TypeBase):
# Schedule control
next_run_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime, nullable=False, server_default=func.current_timestamp(), init=False
)
created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=func.current_timestamp())
updated_at: Mapped[datetime] = mapped_column(
DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp(), init=False
DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
)
def to_dict(self) -> dict[str, Any]:
+2
View File
@@ -895,6 +895,8 @@ class WorkflowNodeExecutionModel(Base): # This model is expected to have `offlo
extras: dict[str, Any] = {}
if self.execution_metadata_dict:
from core.workflow.nodes import NodeType
if self.node_type == NodeType.TOOL and "tool_info" in self.execution_metadata_dict:
tool_info: dict[str, Any] = self.execution_metadata_dict["tool_info"]
extras["icon"] = ToolManager.get_tool_icon(
+3 -3
View File
@@ -1,6 +1,6 @@
[project]
name = "dify-api"
version = "1.10.0"
version = "1.9.2"
requires-python = ">=3.11,<3.13"
dependencies = [
@@ -37,7 +37,7 @@ dependencies = [
"numpy~=1.26.4",
"openpyxl~=3.1.5",
"opik~=1.8.72",
"litellm==1.77.1", # Pinned to avoid madoka dependency issue
"litellm==1.77.1", # Pinned to avoid madoka dependency issue
"opentelemetry-api==1.27.0",
"opentelemetry-distro==0.48b0",
"opentelemetry-exporter-otlp==1.27.0",
@@ -79,6 +79,7 @@ dependencies = [
"tiktoken~=0.9.0",
"transformers~=4.56.1",
"unstructured[docx,epub,md,ppt,pptx]~=0.16.1",
"weave~=0.51.0",
"yarl~=1.18.3",
"webvtt-py~=0.5.1",
"sseclient-py~=1.8.0",
@@ -89,7 +90,6 @@ dependencies = [
"croniter>=6.0.0",
"weaviate-client==4.17.0",
"apscheduler>=3.11.0",
"weave>=0.52.16",
]
# Before adding new dependency, consider place it in
# alphabet order (a-z) and suitable group.
+19 -8
View File
@@ -9,6 +9,7 @@ from extensions.ext_database import db
from libs.datetime_utils import naive_utc_now
from libs.schedule_utils import calculate_next_run_at
from models.trigger import AppTrigger, AppTriggerStatus, AppTriggerType, WorkflowSchedulePlan
from services.workflow.queue_dispatcher import QueueDispatcherManager
from tasks.workflow_schedule_tasks import run_schedule_trigger
logger = logging.getLogger(__name__)
@@ -28,6 +29,7 @@ def poll_workflow_schedules() -> None:
with session_factory() as session:
total_dispatched = 0
total_rate_limited = 0
# Process in batches until we've handled all due schedules or hit the limit
while True:
@@ -36,10 +38,11 @@ def poll_workflow_schedules() -> None:
if not due_schedules:
break
dispatched_count = _process_schedules(session, due_schedules)
dispatched_count, rate_limited_count = _process_schedules(session, due_schedules)
total_dispatched += dispatched_count
total_rate_limited += rate_limited_count
logger.debug("Batch processed: %d dispatched", dispatched_count)
logger.debug("Batch processed: %d dispatched, %d rate limited", dispatched_count, rate_limited_count)
# Circuit breaker: check if we've hit the per-tick limit (if enabled)
if (
@@ -52,8 +55,8 @@ def poll_workflow_schedules() -> None:
)
break
if total_dispatched > 0:
logger.info("Total processed: %d dispatched", total_dispatched)
if total_dispatched > 0 or total_rate_limited > 0:
logger.info("Total processed: %d dispatched, %d rate limited", total_dispatched, total_rate_limited)
def _fetch_due_schedules(session: Session) -> list[WorkflowSchedulePlan]:
@@ -90,12 +93,15 @@ def _fetch_due_schedules(session: Session) -> list[WorkflowSchedulePlan]:
return list(due_schedules)
def _process_schedules(session: Session, schedules: list[WorkflowSchedulePlan]) -> int:
def _process_schedules(session: Session, schedules: list[WorkflowSchedulePlan]) -> tuple[int, int]:
"""Process schedules: check quota, update next run time and dispatch to Celery in parallel."""
if not schedules:
return 0
return 0, 0
dispatcher_manager = QueueDispatcherManager()
tasks_to_dispatch: list[str] = []
rate_limited_count = 0
for schedule in schedules:
next_run_at = calculate_next_run_at(
schedule.cron_expression,
@@ -103,7 +109,12 @@ def _process_schedules(session: Session, schedules: list[WorkflowSchedulePlan])
)
schedule.next_run_at = next_run_at
tasks_to_dispatch.append(schedule.id)
dispatcher = dispatcher_manager.get_dispatcher(schedule.tenant_id)
if not dispatcher.check_daily_quota(schedule.tenant_id):
logger.info("Tenant %s rate limited, skipping schedule_plan %s", schedule.tenant_id, schedule.id)
rate_limited_count += 1
else:
tasks_to_dispatch.append(schedule.id)
if tasks_to_dispatch:
job = group(run_schedule_trigger.s(schedule_id) for schedule_id in tasks_to_dispatch)
@@ -113,4 +124,4 @@ def _process_schedules(session: Session, schedules: list[WorkflowSchedulePlan])
session.commit()
return len(tasks_to_dispatch)
return len(tasks_to_dispatch), rate_limited_count
+17 -8
View File
@@ -10,14 +10,19 @@ from core.app.apps.completion.app_generator import CompletionAppGenerator
from core.app.apps.workflow.app_generator import WorkflowAppGenerator
from core.app.entities.app_invoke_entities import InvokeFrom
from core.app.features.rate_limiting import RateLimit
from enums.quota_type import QuotaType, unlimited
from enums.cloud_plan import CloudPlan
from libs.helper import RateLimiter
from models.model import Account, App, AppMode, EndUser
from models.workflow import Workflow
from services.errors.app import InvokeRateLimitError, QuotaExceededError, WorkflowIdFormatError, WorkflowNotFoundError
from services.billing_service import BillingService
from services.errors.app import WorkflowIdFormatError, WorkflowNotFoundError
from services.errors.llm import InvokeRateLimitError
from services.workflow_service import WorkflowService
class AppGenerateService:
system_rate_limiter = RateLimiter("app_daily_rate_limiter", dify_config.APP_DAILY_RATE_LIMIT, 86400)
@classmethod
def generate(
cls,
@@ -37,12 +42,17 @@ class AppGenerateService:
:param streaming: streaming
:return:
"""
quota_charge = unlimited()
# system level rate limiter
if dify_config.BILLING_ENABLED:
try:
quota_charge = QuotaType.WORKFLOW.consume(app_model.tenant_id)
except QuotaExceededError:
raise InvokeRateLimitError(f"Workflow execution quota limit reached for tenant {app_model.tenant_id}")
# check if it's free plan
limit_info = BillingService.get_info(app_model.tenant_id)
if limit_info["subscription"]["plan"] == CloudPlan.SANDBOX:
if cls.system_rate_limiter.is_rate_limited(app_model.tenant_id):
raise InvokeRateLimitError(
"Rate limit exceeded, please upgrade your plan "
f"or your RPD was {dify_config.APP_DAILY_RATE_LIMIT} requests/day"
)
cls.system_rate_limiter.increment_rate_limit(app_model.tenant_id)
# app level rate limiter
max_active_request = cls._get_max_active_requests(app_model)
@@ -114,7 +124,6 @@ class AppGenerateService:
else:
raise ValueError(f"Invalid app mode {app_model.mode}")
except Exception:
quota_charge.refund()
rate_limit.exit(request_id)
raise
finally:
+18 -10
View File
@@ -13,17 +13,18 @@ from celery.result import AsyncResult
from sqlalchemy import select
from sqlalchemy.orm import Session
from enums.quota_type import QuotaType
from extensions.ext_database import db
from extensions.ext_redis import redis_client
from models.account import Account
from models.enums import CreatorUserRole, WorkflowTriggerStatus
from models.model import App, EndUser
from models.trigger import WorkflowTriggerLog
from models.workflow import Workflow
from repositories.sqlalchemy_workflow_trigger_log_repository import SQLAlchemyWorkflowTriggerLogRepository
from services.errors.app import InvokeRateLimitError, QuotaExceededError, WorkflowNotFoundError
from services.errors.app import InvokeDailyRateLimitError, WorkflowNotFoundError
from services.workflow.entities import AsyncTriggerResponse, TriggerData, WorkflowTaskData
from services.workflow.queue_dispatcher import QueueDispatcherManager, QueuePriority
from services.workflow.rate_limiter import TenantDailyRateLimiter
from services.workflow_service import WorkflowService
from tasks.async_workflow_tasks import (
execute_workflow_professional,
@@ -81,6 +82,7 @@ class AsyncWorkflowService:
trigger_log_repo = SQLAlchemyWorkflowTriggerLogRepository(session)
dispatcher_manager = QueueDispatcherManager()
workflow_service = WorkflowService()
rate_limiter = TenantDailyRateLimiter(redis_client)
# 1. Validate app exists
app_model = session.scalar(select(App).where(App.id == trigger_data.app_id))
@@ -125,19 +127,25 @@ class AsyncWorkflowService:
trigger_log = trigger_log_repo.create(trigger_log)
session.commit()
# 7. Check and consume quota
try:
QuotaType.WORKFLOW.consume(trigger_data.tenant_id)
except QuotaExceededError as e:
# 7. Check and consume daily quota
if not dispatcher.consume_quota(trigger_data.tenant_id):
# Update trigger log status
trigger_log.status = WorkflowTriggerStatus.RATE_LIMITED
trigger_log.error = f"Quota limit reached: {e}"
trigger_log.error = f"Daily limit reached for {dispatcher.get_queue_name()}"
trigger_log_repo.update(trigger_log)
session.commit()
raise InvokeRateLimitError(
f"Workflow execution quota limit reached for tenant {trigger_data.tenant_id}"
) from e
tenant_owner_tz = rate_limiter.get_tenant_owner_timezone(trigger_data.tenant_id)
remaining = rate_limiter.get_remaining_quota(trigger_data.tenant_id, dispatcher.get_daily_limit())
reset_time = rate_limiter.get_quota_reset_time(trigger_data.tenant_id, tenant_owner_tz)
raise InvokeDailyRateLimitError(
f"Daily workflow execution limit reached. "
f"Limit resets at {reset_time.strftime('%Y-%m-%d %H:%M:%S %Z')}. "
f"Remaining quota: {remaining}"
)
# 8. Create task data
queue_name = dispatcher.get_queue_name()
-47
View File
@@ -24,13 +24,6 @@ class BillingService:
billing_info = cls._send_request("GET", "/subscription/info", params=params)
return billing_info
@classmethod
def get_tenant_feature_plan_usage_info(cls, tenant_id: str):
params = {"tenant_id": tenant_id}
usage_info = cls._send_request("GET", "/tenant-feature-usage/info", params=params)
return usage_info
@classmethod
def get_knowledge_rate_limit(cls, tenant_id: str):
params = {"tenant_id": tenant_id}
@@ -62,44 +55,6 @@ class BillingService:
params = {"prefilled_email": prefilled_email, "tenant_id": tenant_id}
return cls._send_request("GET", "/invoices", params=params)
@classmethod
def update_tenant_feature_plan_usage(cls, tenant_id: str, feature_key: str, delta: int) -> dict:
"""
Update tenant feature plan usage.
Args:
tenant_id: Tenant identifier
feature_key: Feature key (e.g., 'trigger', 'workflow')
delta: Usage delta (positive to add, negative to consume)
Returns:
Response dict with 'result' and 'history_id'
Example: {"result": "success", "history_id": "uuid"}
"""
return cls._send_request(
"POST",
"/tenant-feature-usage/usage",
params={"tenant_id": tenant_id, "feature_key": feature_key, "delta": delta},
)
@classmethod
def refund_tenant_feature_plan_usage(cls, history_id: str) -> dict:
"""
Refund a previous usage charge.
Args:
history_id: The history_id returned from update_tenant_feature_plan_usage
Returns:
Response dict with 'result' and 'history_id'
"""
return cls._send_request("POST", "/tenant-feature-usage/refund", params={"quota_usage_history_id": history_id})
@classmethod
def get_tenant_feature_plan_usage(cls, tenant_id: str, feature_key: str):
params = {"tenant_id": tenant_id, "feature_key": feature_key}
return cls._send_request("GET", "/billing/tenant_feature_plan/usage", params=params)
@classmethod
@retry(
wait=wait_fixed(2),
@@ -114,8 +69,6 @@ class BillingService:
response = httpx.request(method, url, json=json, params=params, headers=headers)
if method == "GET" and response.status_code != httpx.codes.OK:
raise ValueError("Unable to retrieve billing information. Please try again later or contact support.")
if method == "POST" and response.status_code != httpx.codes.OK:
raise ValueError(f"Unable to send request to {url}. Please try again later or contact support.")
return response.json()
@staticmethod
-56
View File
@@ -1082,62 +1082,6 @@ class DocumentService:
},
}
DISPLAY_STATUS_ALIASES: dict[str, str] = {
"active": "available",
"enabled": "available",
}
_INDEXING_STATUSES: tuple[str, ...] = ("parsing", "cleaning", "splitting", "indexing")
DISPLAY_STATUS_FILTERS: dict[str, tuple[Any, ...]] = {
"queuing": (Document.indexing_status == "waiting",),
"indexing": (
Document.indexing_status.in_(_INDEXING_STATUSES),
Document.is_paused.is_not(True),
),
"paused": (
Document.indexing_status.in_(_INDEXING_STATUSES),
Document.is_paused.is_(True),
),
"error": (Document.indexing_status == "error",),
"available": (
Document.indexing_status == "completed",
Document.archived.is_(False),
Document.enabled.is_(True),
),
"disabled": (
Document.indexing_status == "completed",
Document.archived.is_(False),
Document.enabled.is_(False),
),
"archived": (
Document.indexing_status == "completed",
Document.archived.is_(True),
),
}
@classmethod
def normalize_display_status(cls, status: str | None) -> str | None:
if not status:
return None
normalized = status.lower()
normalized = cls.DISPLAY_STATUS_ALIASES.get(normalized, normalized)
return normalized if normalized in cls.DISPLAY_STATUS_FILTERS else None
@classmethod
def build_display_status_filters(cls, status: str | None) -> tuple[Any, ...]:
normalized = cls.normalize_display_status(status)
if not normalized:
return ()
return cls.DISPLAY_STATUS_FILTERS[normalized]
@classmethod
def apply_display_status_filter(cls, query, status: str | None):
filters = cls.build_display_status_filters(status)
if not filters:
return query
return query.where(*filters)
DOCUMENT_METADATA_SCHEMA: dict[str, Any] = {
"book": {
"title": str,
+2 -24
View File
@@ -18,29 +18,7 @@ class WorkflowIdFormatError(Exception):
pass
class InvokeRateLimitError(Exception):
"""Raised when rate limit is exceeded for workflow invocations."""
class InvokeDailyRateLimitError(Exception):
"""Raised when daily rate limit is exceeded for workflow invocations."""
pass
class QuotaExceededError(ValueError):
"""Raised when billing quota is exceeded for a feature."""
def __init__(self, feature: str, tenant_id: str, required: int):
self.feature = feature
self.tenant_id = tenant_id
self.required = required
super().__init__(f"Quota exceeded for feature '{feature}' (tenant: {tenant_id}). Required: {required}")
class TriggerNodeLimitExceededError(ValueError):
"""Raised when trigger node count exceeds the plan limit."""
def __init__(self, count: int, limit: int):
self.count = count
self.limit = limit
super().__init__(
f"Trigger node count ({count}) exceeds the limit ({limit}) for your subscription plan. "
f"Please upgrade your plan or reduce the number of trigger nodes."
)
+3 -3
View File
@@ -62,7 +62,7 @@ class ExternalDatasetService:
tenant_id=tenant_id,
created_by=user_id,
updated_by=user_id,
name=str(args.get("name")),
name=args.get("name"),
description=args.get("description", ""),
settings=json.dumps(args.get("settings"), ensure_ascii=False),
)
@@ -163,7 +163,7 @@ class ExternalDatasetService:
external_knowledge_api = (
db.session.query(ExternalKnowledgeApis).filter_by(id=external_knowledge_api_id, tenant_id=tenant_id).first()
)
if external_knowledge_api is None or external_knowledge_api.settings is None:
if external_knowledge_api is None:
raise ValueError("api template not found")
settings = json.loads(external_knowledge_api.settings)
for setting in settings:
@@ -290,7 +290,7 @@ class ExternalDatasetService:
.filter_by(id=external_knowledge_binding.external_knowledge_api_id)
.first()
)
if external_knowledge_api is None or external_knowledge_api.settings is None:
if not external_knowledge_api:
raise ValueError("external api template not found")
settings = json.loads(external_knowledge_api.settings)
-20
View File
@@ -54,12 +54,6 @@ class LicenseLimitationModel(BaseModel):
return (self.limit - self.size) >= required
class Quota(BaseModel):
usage: int = 0
limit: int = 0
reset_date: int = -1
class LicenseStatus(StrEnum):
NONE = "none"
INACTIVE = "inactive"
@@ -135,8 +129,6 @@ class FeatureModel(BaseModel):
webapp_copyright_enabled: bool = False
workspace_members: LicenseLimitationModel = LicenseLimitationModel(enabled=False, size=0, limit=0)
is_allow_transfer_workspace: bool = True
trigger_event: Quota = Quota(usage=0, limit=3000, reset_date=0)
api_rate_limit: Quota = Quota(usage=0, limit=5000, reset_date=0)
# pydantic configs
model_config = ConfigDict(protected_namespaces=())
knowledge_pipeline: KnowledgePipeline = KnowledgePipeline()
@@ -244,8 +236,6 @@ class FeatureService:
def _fulfill_params_from_billing_api(cls, features: FeatureModel, tenant_id: str):
billing_info = BillingService.get_info(tenant_id)
features_usage_info = BillingService.get_tenant_feature_plan_usage_info(tenant_id)
features.billing.enabled = billing_info["enabled"]
features.billing.subscription.plan = billing_info["subscription"]["plan"]
features.billing.subscription.interval = billing_info["subscription"]["interval"]
@@ -256,16 +246,6 @@ class FeatureService:
else:
features.is_allow_transfer_workspace = False
if "trigger_event" in features_usage_info:
features.trigger_event.usage = features_usage_info["trigger_event"]["usage"]
features.trigger_event.limit = features_usage_info["trigger_event"]["limit"]
features.trigger_event.reset_date = features_usage_info["trigger_event"].get("reset_date", -1)
if "api_rate_limit" in features_usage_info:
features.api_rate_limit.usage = features_usage_info["api_rate_limit"]["usage"]
features.api_rate_limit.limit = features_usage_info["api_rate_limit"]["limit"]
features.api_rate_limit.reset_date = features_usage_info["api_rate_limit"].get("reset_date", -1)
if "members" in billing_info:
features.members.size = billing_info["members"]["size"]
features.members.limit = billing_info["members"]["limit"]
+2 -13
View File
@@ -507,11 +507,7 @@ class MCPToolManageService:
return auth_result.response
def auth_with_actions(
self,
provider_entity: MCPProviderEntity,
authorization_code: str | None = None,
resource_metadata_url: str | None = None,
scope_hint: str | None = None,
self, provider_entity: MCPProviderEntity, authorization_code: str | None = None
) -> dict[str, str]:
"""
Perform authentication and execute all resulting actions.
@@ -521,18 +517,11 @@ class MCPToolManageService:
Args:
provider_entity: The MCP provider entity
authorization_code: Optional authorization code
resource_metadata_url: Optional Protected Resource Metadata URL from WWW-Authenticate
scope_hint: Optional scope hint from WWW-Authenticate header
Returns:
Response dictionary from auth result
"""
auth_result = auth(
provider_entity,
authorization_code,
resource_metadata_url=resource_metadata_url,
scope_hint=scope_hint,
)
auth_result = auth(provider_entity, authorization_code)
return self.execute_auth_actions(auth_result)
def _reconnect_provider(self, *, server_url: str, provider: MCPToolProvider) -> ReconnectResult:
@@ -1,46 +0,0 @@
"""
AppTrigger management service.
Handles AppTrigger model CRUD operations and status management.
This service centralizes all AppTrigger-related business logic.
"""
import logging
from sqlalchemy import update
from sqlalchemy.orm import Session
from extensions.ext_database import db
from models.enums import AppTriggerStatus
from models.trigger import AppTrigger
logger = logging.getLogger(__name__)
class AppTriggerService:
"""Service for managing AppTrigger lifecycle and status."""
@staticmethod
def mark_tenant_triggers_rate_limited(tenant_id: str) -> None:
"""
Mark all enabled triggers for a tenant as rate limited due to quota exceeded.
This method is called when a tenant's quota is exhausted. It updates all
enabled triggers to RATE_LIMITED status to prevent further executions until
quota is restored.
Args:
tenant_id: Tenant ID whose triggers should be marked as rate limited
"""
try:
with Session(db.engine) as session:
session.execute(
update(AppTrigger)
.where(AppTrigger.tenant_id == tenant_id, AppTrigger.status == AppTriggerStatus.ENABLED)
.values(status=AppTriggerStatus.RATE_LIMITED)
)
session.commit()
logger.info("Marked all enabled triggers as rate limited for tenant %s", tenant_id)
except Exception:
logger.exception("Failed to mark all enabled triggers as rate limited for tenant %s", tenant_id)
+3
View File
@@ -298,6 +298,9 @@ class TriggerService:
redis_client.delete(f"{cls.__PLUGIN_TRIGGER_NODE_CACHE_KEY__}:{node_id}")
session.commit()
except Exception:
import logging
logger = logging.getLogger(__name__)
logger.exception("Failed to sync plugin trigger relationships for app %s", app.id)
raise
finally:
-21
View File
@@ -18,7 +18,6 @@ from core.file.models import FileTransferMethod
from core.tools.tool_file_manager import ToolFileManager
from core.variables.types import SegmentType
from core.workflow.enums import NodeType
from enums.quota_type import QuotaType
from extensions.ext_database import db
from extensions.ext_redis import redis_client
from factories import file_factory
@@ -28,8 +27,6 @@ from models.trigger import AppTrigger, WorkflowWebhookTrigger
from models.workflow import Workflow
from services.async_workflow_service import AsyncWorkflowService
from services.end_user_service import EndUserService
from services.errors.app import QuotaExceededError
from services.trigger.app_trigger_service import AppTriggerService
from services.workflow.entities import WebhookTriggerData
logger = logging.getLogger(__name__)
@@ -101,12 +98,6 @@ class WebhookService:
raise ValueError(f"App trigger not found for webhook {webhook_id}")
# Only check enabled status if not in debug mode
if app_trigger.status == AppTriggerStatus.RATE_LIMITED:
raise ValueError(
f"Webhook trigger is rate limited for webhook {webhook_id}, please upgrade your plan."
)
if app_trigger.status != AppTriggerStatus.ENABLED:
raise ValueError(f"Webhook trigger is disabled for webhook {webhook_id}")
@@ -738,18 +729,6 @@ class WebhookService:
user_id=None,
)
# consume quota before triggering workflow execution
try:
QuotaType.TRIGGER.consume(webhook_trigger.tenant_id)
except QuotaExceededError:
AppTriggerService.mark_tenant_triggers_rate_limited(webhook_trigger.tenant_id)
logger.info(
"Tenant %s rate limited, skipping webhook trigger %s",
webhook_trigger.tenant_id,
webhook_trigger.webhook_id,
)
raise
# Trigger workflow execution asynchronously
AsyncWorkflowService.trigger_workflow_async(
session,
+46 -1
View File
@@ -2,14 +2,16 @@
Queue dispatcher system for async workflow execution.
Implements an ABC-based pattern for handling different subscription tiers
with appropriate queue routing and priority assignment.
with appropriate queue routing and rate limiting.
"""
from abc import ABC, abstractmethod
from enum import StrEnum
from configs import dify_config
from extensions.ext_redis import redis_client
from services.billing_service import BillingService
from services.workflow.rate_limiter import TenantDailyRateLimiter
class QueuePriority(StrEnum):
@@ -23,16 +25,50 @@ class QueuePriority(StrEnum):
class BaseQueueDispatcher(ABC):
"""Abstract base class for queue dispatchers"""
def __init__(self):
self.rate_limiter = TenantDailyRateLimiter(redis_client)
@abstractmethod
def get_queue_name(self) -> str:
"""Get the queue name for this dispatcher"""
pass
@abstractmethod
def get_daily_limit(self) -> int:
"""Get daily execution limit"""
pass
@abstractmethod
def get_priority(self) -> int:
"""Get task priority level"""
pass
def check_daily_quota(self, tenant_id: str) -> bool:
"""
Check if tenant has remaining daily quota
Args:
tenant_id: The tenant identifier
Returns:
True if quota available, False otherwise
"""
# Check without consuming
remaining = self.rate_limiter.get_remaining_quota(tenant_id=tenant_id, max_daily_limit=self.get_daily_limit())
return remaining > 0
def consume_quota(self, tenant_id: str) -> bool:
"""
Consume one execution from daily quota
Args:
tenant_id: The tenant identifier
Returns:
True if quota consumed successfully, False if limit reached
"""
return self.rate_limiter.check_and_consume(tenant_id=tenant_id, max_daily_limit=self.get_daily_limit())
class ProfessionalQueueDispatcher(BaseQueueDispatcher):
"""Dispatcher for professional tier"""
@@ -40,6 +76,9 @@ class ProfessionalQueueDispatcher(BaseQueueDispatcher):
def get_queue_name(self) -> str:
return QueuePriority.PROFESSIONAL
def get_daily_limit(self) -> int:
return int(1e9)
def get_priority(self) -> int:
return 100
@@ -50,6 +89,9 @@ class TeamQueueDispatcher(BaseQueueDispatcher):
def get_queue_name(self) -> str:
return QueuePriority.TEAM
def get_daily_limit(self) -> int:
return int(1e9)
def get_priority(self) -> int:
return 50
@@ -60,6 +102,9 @@ class SandboxQueueDispatcher(BaseQueueDispatcher):
def get_queue_name(self) -> str:
return QueuePriority.SANDBOX
def get_daily_limit(self) -> int:
return dify_config.APP_DAILY_RATE_LIMIT
def get_priority(self) -> int:
return 10
+183
View File
@@ -0,0 +1,183 @@
"""
Day-based rate limiter for workflow executions.
Implements UTC-based daily quotas that reset at midnight UTC for consistent rate limiting.
"""
from datetime import UTC, datetime, time, timedelta
from typing import Union
import pytz
from redis import Redis
from sqlalchemy import select
from extensions.ext_database import db
from extensions.ext_redis import RedisClientWrapper
from models.account import Account, TenantAccountJoin, TenantAccountRole
class TenantDailyRateLimiter:
"""
Day-based rate limiter that resets at midnight UTC
This class provides Redis-based rate limiting with the following features:
- Daily quotas that reset at midnight UTC for consistency
- Atomic check-and-consume operations
- Automatic cleanup of stale counters
- Timezone-aware error messages for better UX
"""
def __init__(self, redis_client: Union[Redis, RedisClientWrapper]):
self.redis = redis_client
def get_tenant_owner_timezone(self, tenant_id: str) -> str:
"""
Get timezone of tenant owner
Args:
tenant_id: The tenant identifier
Returns:
Timezone string (e.g., 'America/New_York', 'UTC')
"""
# Query to get tenant owner's timezone using scalar and select
owner = db.session.scalar(
select(Account)
.join(TenantAccountJoin, TenantAccountJoin.account_id == Account.id)
.where(TenantAccountJoin.tenant_id == tenant_id, TenantAccountJoin.role == TenantAccountRole.OWNER)
)
if not owner:
return "UTC"
return owner.timezone or "UTC"
def _get_day_key(self, tenant_id: str) -> str:
"""
Get Redis key for current UTC day
Args:
tenant_id: The tenant identifier
Returns:
Redis key for the current UTC day
"""
utc_now = datetime.now(UTC)
date_str = utc_now.strftime("%Y-%m-%d")
return f"workflow:daily_limit:{tenant_id}:{date_str}"
def _get_ttl_seconds(self) -> int:
"""
Calculate seconds until UTC midnight
Returns:
Number of seconds until UTC midnight
"""
utc_now = datetime.now(UTC)
# Get next midnight in UTC
next_midnight = datetime.combine(utc_now.date() + timedelta(days=1), time.min)
next_midnight = next_midnight.replace(tzinfo=UTC)
return int((next_midnight - utc_now).total_seconds())
def check_and_consume(self, tenant_id: str, max_daily_limit: int) -> bool:
"""
Check if quota available and consume one execution
Args:
tenant_id: The tenant identifier
max_daily_limit: Maximum daily limit
Returns:
True if quota consumed successfully, False if limit reached
"""
key = self._get_day_key(tenant_id)
ttl = self._get_ttl_seconds()
# Check current usage
current = self.redis.get(key)
if current is None:
# First execution of the day - set to 1
self.redis.setex(key, ttl, 1)
return True
current_count = int(current)
if current_count < max_daily_limit:
# Within limit, increment
new_count = self.redis.incr(key)
# Update TTL
self.redis.expire(key, ttl)
# Double-check in case of race condition
if new_count <= max_daily_limit:
return True
else:
# Race condition occurred, decrement back
self.redis.decr(key)
return False
else:
# Limit exceeded
return False
def get_remaining_quota(self, tenant_id: str, max_daily_limit: int) -> int:
"""
Get remaining quota for the day
Args:
tenant_id: The tenant identifier
max_daily_limit: Maximum daily limit
Returns:
Number of remaining executions for the day
"""
key = self._get_day_key(tenant_id)
used = int(self.redis.get(key) or 0)
return max(0, max_daily_limit - used)
def get_current_usage(self, tenant_id: str) -> int:
"""
Get current usage for the day
Args:
tenant_id: The tenant identifier
Returns:
Number of executions used today
"""
key = self._get_day_key(tenant_id)
return int(self.redis.get(key) or 0)
def reset_quota(self, tenant_id: str) -> bool:
"""
Reset quota for testing purposes
Args:
tenant_id: The tenant identifier
Returns:
True if key was deleted, False if key didn't exist
"""
key = self._get_day_key(tenant_id)
return bool(self.redis.delete(key))
def get_quota_reset_time(self, tenant_id: str, timezone_str: str) -> datetime:
"""
Get the time when quota will reset (next UTC midnight in tenant's timezone)
Args:
tenant_id: The tenant identifier
timezone_str: Tenant's timezone for display purposes
Returns:
Datetime when quota resets (next UTC midnight in tenant's timezone)
"""
tz = pytz.timezone(timezone_str)
utc_now = datetime.now(UTC)
# Get next midnight in UTC, then convert to tenant's timezone
next_utc_midnight = datetime.combine(utc_now.date() + timedelta(days=1), time.min)
next_utc_midnight = pytz.UTC.localize(next_utc_midnight)
return next_utc_midnight.astimezone(tz)
+1 -19
View File
@@ -7,7 +7,6 @@ from typing import Any, cast
from sqlalchemy import exists, select
from sqlalchemy.orm import Session, sessionmaker
from configs import dify_config
from core.app.app_config.entities import VariableEntityType
from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfigManager
from core.app.apps.workflow.app_config_manager import WorkflowAppConfigManager
@@ -26,7 +25,6 @@ from core.workflow.nodes.node_mapping import LATEST_VERSION, NODE_TYPE_CLASSES_M
from core.workflow.nodes.start.entities import StartNodeData
from core.workflow.system_variable import SystemVariable
from core.workflow.workflow_entry import WorkflowEntry
from enums.cloud_plan import CloudPlan
from events.app_event import app_draft_workflow_was_synced, app_published_workflow_was_updated
from extensions.ext_database import db
from extensions.ext_storage import storage
@@ -37,9 +35,8 @@ from models.model import App, AppMode
from models.tools import WorkflowToolProvider
from models.workflow import Workflow, WorkflowNodeExecutionModel, WorkflowNodeExecutionTriggeredFrom, WorkflowType
from repositories.factory import DifyAPIRepositoryFactory
from services.billing_service import BillingService
from services.enterprise.plugin_manager_service import PluginCredentialType
from services.errors.app import IsDraftWorkflowError, TriggerNodeLimitExceededError, WorkflowHashNotEqualError
from services.errors.app import IsDraftWorkflowError, WorkflowHashNotEqualError
from services.workflow.workflow_converter import WorkflowConverter
from .errors.workflow_service import DraftWorkflowDeletionError, WorkflowInUseError
@@ -275,21 +272,6 @@ class WorkflowService:
# validate graph structure
self.validate_graph_structure(graph=draft_workflow.graph_dict)
# billing check
if dify_config.BILLING_ENABLED:
limit_info = BillingService.get_info(app_model.tenant_id)
if limit_info["subscription"]["plan"] == CloudPlan.SANDBOX:
# Check trigger node count limit for SANDBOX plan
trigger_node_count = sum(
1
for _, node_data in draft_workflow.walk_nodes()
if (node_type_str := node_data.get("type"))
and isinstance(node_type_str, str)
and NodeType(node_type_str).is_trigger_node
)
if trigger_node_count > 2:
raise TriggerNodeLimitExceededError(count=trigger_node_count, limit=2)
# create new workflow
workflow = Workflow.new(
tenant_id=app_model.tenant_id,
+2 -13
View File
@@ -13,7 +13,7 @@ from sqlalchemy import select
from sqlalchemy.orm import Session, sessionmaker
from configs import dify_config
from core.app.apps.workflow.app_generator import SKIP_PREPARE_USER_INPUTS_KEY, WorkflowAppGenerator
from core.app.apps.workflow.app_generator import WorkflowAppGenerator
from core.app.entities.app_invoke_entities import InvokeFrom
from core.app.layers.timeslice_layer import TimeSliceLayer
from core.app.layers.trigger_post_layer import TriggerPostLayer
@@ -81,17 +81,6 @@ def execute_workflow_sandbox(task_data_dict: dict[str, Any]):
)
def _build_generator_args(trigger_data: TriggerData) -> dict[str, Any]:
"""Build args passed into WorkflowAppGenerator.generate for Celery executions."""
args: dict[str, Any] = {
"inputs": dict(trigger_data.inputs),
"files": list(trigger_data.files),
SKIP_PREPARE_USER_INPUTS_KEY: True,
}
return args
def _execute_workflow_common(
task_data: WorkflowTaskData,
cfs_plan_scheduler: AsyncWorkflowCFSPlanScheduler,
@@ -139,7 +128,7 @@ def _execute_workflow_common(
generator = WorkflowAppGenerator()
# Prepare args matching AppGenerateService.generate format
args = _build_generator_args(trigger_data)
args: dict[str, Any] = {"inputs": dict(trigger_data.inputs), "files": list(trigger_data.files)}
# If workflow_id was specified, add it to args
if trigger_data.workflow_id:
+2 -8
View File
@@ -9,7 +9,7 @@ from core.rag.index_processor.index_processor_factory import IndexProcessorFacto
from core.tools.utils.web_reader_tool import get_image_upload_file_ids
from extensions.ext_database import db
from extensions.ext_storage import storage
from models.dataset import Dataset, DatasetMetadataBinding, DocumentSegment
from models.dataset import Dataset, DocumentSegment
from models.model import UploadFile
logger = logging.getLogger(__name__)
@@ -37,11 +37,6 @@ def batch_clean_document_task(document_ids: list[str], dataset_id: str, doc_form
if not dataset:
raise Exception("Document has no dataset")
db.session.query(DatasetMetadataBinding).where(
DatasetMetadataBinding.dataset_id == dataset_id,
DatasetMetadataBinding.document_id.in_(document_ids),
).delete(synchronize_session=False)
segments = db.session.scalars(
select(DocumentSegment).where(DocumentSegment.document_id.in_(document_ids))
).all()
@@ -76,8 +71,7 @@ def batch_clean_document_task(document_ids: list[str], dataset_id: str, doc_form
except Exception:
logger.exception("Delete file failed when document deleted, file_id: %s", file.id)
db.session.delete(file)
db.session.commit()
db.session.commit()
end_at = time.perf_counter()
logger.info(
+1 -28
View File
@@ -26,22 +26,14 @@ from core.trigger.provider import PluginTriggerProviderController
from core.trigger.trigger_manager import TriggerManager
from core.workflow.enums import NodeType, WorkflowExecutionStatus
from core.workflow.nodes.trigger_plugin.entities import TriggerEventNodeData
from enums.quota_type import QuotaType, unlimited
from extensions.ext_database import db
from models.enums import (
AppTriggerType,
CreatorUserRole,
WorkflowRunTriggeredFrom,
WorkflowTriggerStatus,
)
from models.enums import AppTriggerType, CreatorUserRole, WorkflowRunTriggeredFrom, WorkflowTriggerStatus
from models.model import EndUser
from models.provider_ids import TriggerProviderID
from models.trigger import TriggerSubscription, WorkflowPluginTrigger, WorkflowTriggerLog
from models.workflow import Workflow, WorkflowAppLog, WorkflowAppLogCreatedFrom, WorkflowRun
from services.async_workflow_service import AsyncWorkflowService
from services.end_user_service import EndUserService
from services.errors.app import QuotaExceededError
from services.trigger.app_trigger_service import AppTriggerService
from services.trigger.trigger_provider_service import TriggerProviderService
from services.trigger.trigger_request_service import TriggerHttpRequestCachingService
from services.trigger.trigger_subscription_operator_service import TriggerSubscriptionOperatorService
@@ -295,17 +287,6 @@ def dispatch_triggered_workflow(
icon_dark_filename=trigger_entity.identity.icon_dark or "",
)
# consume quota before invoking trigger
quota_charge = unlimited()
try:
quota_charge = QuotaType.TRIGGER.consume(subscription.tenant_id)
except QuotaExceededError:
AppTriggerService.mark_tenant_triggers_rate_limited(subscription.tenant_id)
logger.info(
"Tenant %s rate limited, skipping plugin trigger %s", subscription.tenant_id, plugin_trigger.id
)
return 0
node_data: TriggerEventNodeData = TriggerEventNodeData.model_validate(event_node)
invoke_response: TriggerInvokeEventResponse | None = None
try:
@@ -324,8 +305,6 @@ def dispatch_triggered_workflow(
payload=payload,
)
except PluginInvokeError as e:
quota_charge.refund()
error_message = e.to_user_friendly_error(plugin_name=trigger_entity.identity.name)
try:
end_user = end_users.get(plugin_trigger.app_id)
@@ -347,8 +326,6 @@ def dispatch_triggered_workflow(
)
continue
except Exception:
quota_charge.refund()
logger.exception(
"Failed to invoke trigger event for app %s",
plugin_trigger.app_id,
@@ -356,8 +333,6 @@ def dispatch_triggered_workflow(
continue
if invoke_response is not None and invoke_response.cancelled:
quota_charge.refund()
logger.info(
"Trigger ignored for app %s with trigger event %s",
plugin_trigger.app_id,
@@ -391,8 +366,6 @@ def dispatch_triggered_workflow(
event_name,
)
except Exception:
quota_charge.refund()
logger.exception(
"Failed to trigger workflow for app %s",
plugin_trigger.app_id,
@@ -6,7 +6,6 @@ from typing import Any
from celery import shared_task
from sqlalchemy.orm import Session
from configs import dify_config
from core.plugin.entities.plugin_daemon import CredentialType
from core.trigger.utils.locks import build_trigger_refresh_lock_key
from extensions.ext_database import db
@@ -26,10 +25,9 @@ def _load_subscription(session: Session, tenant_id: str, subscription_id: str) -
def _refresh_oauth_if_expired(tenant_id: str, subscription: TriggerSubscription, now: int) -> None:
threshold_seconds: int = int(dify_config.TRIGGER_PROVIDER_CREDENTIAL_THRESHOLD_SECONDS)
if (
subscription.credential_expires_at != -1
and int(subscription.credential_expires_at) <= now + threshold_seconds
and int(subscription.credential_expires_at) <= now
and CredentialType.of(subscription.credential_type) == CredentialType.OAUTH2
):
logger.info(
@@ -55,15 +53,13 @@ def _refresh_subscription_if_expired(
subscription: TriggerSubscription,
now: int,
) -> None:
threshold_seconds: int = int(dify_config.TRIGGER_PROVIDER_SUBSCRIPTION_THRESHOLD_SECONDS)
if subscription.expires_at == -1 or int(subscription.expires_at) > now + threshold_seconds:
if subscription.expires_at == -1 or int(subscription.expires_at) > now:
logger.debug(
"Subscription not due: tenant=%s subscription_id=%s expires_at=%s now=%s threshold=%s",
"Subscription not due: tenant=%s subscription_id=%s expires_at=%s now=%s",
tenant_id,
subscription.id,
subscription.expires_at,
now,
threshold_seconds,
)
return
-13
View File
@@ -8,12 +8,9 @@ from core.workflow.nodes.trigger_schedule.exc import (
ScheduleNotFoundError,
TenantOwnerNotFoundError,
)
from enums.quota_type import QuotaType, unlimited
from extensions.ext_database import db
from models.trigger import WorkflowSchedulePlan
from services.async_workflow_service import AsyncWorkflowService
from services.errors.app import QuotaExceededError
from services.trigger.app_trigger_service import AppTriggerService
from services.trigger.schedule_service import ScheduleService
from services.workflow.entities import ScheduleTriggerData
@@ -33,7 +30,6 @@ def run_schedule_trigger(schedule_id: str) -> None:
TenantOwnerNotFoundError: If no owner/admin for tenant
ScheduleExecutionError: If workflow trigger fails
"""
session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
with session_factory() as session:
@@ -45,14 +41,6 @@ def run_schedule_trigger(schedule_id: str) -> None:
if not tenant_owner:
raise TenantOwnerNotFoundError(f"No owner or admin found for tenant {schedule.tenant_id}")
quota_charge = unlimited()
try:
quota_charge = QuotaType.TRIGGER.consume(schedule.tenant_id)
except QuotaExceededError:
AppTriggerService.mark_tenant_triggers_rate_limited(schedule.tenant_id)
logger.info("Tenant %s rate limited, skipping schedule trigger %s", schedule.tenant_id, schedule_id)
return
try:
# Production dispatch: Trigger the workflow normally
response = AsyncWorkflowService.trigger_workflow_async(
@@ -67,7 +55,6 @@ def run_schedule_trigger(schedule_id: str) -> None:
)
logger.info("Schedule %s triggered workflow: %s", schedule_id, response.workflow_trigger_log_id)
except Exception as e:
quota_charge.refund()
raise ScheduleExecutionError(
f"Failed to trigger workflow for schedule {schedule_id}, app {schedule.app_id}"
) from e
@@ -4,6 +4,7 @@ import pymysql
def check_tiflash_ready() -> bool:
connection = None
try:
connection = pymysql.connect(
host="localhost",
@@ -28,7 +29,10 @@ def check_tiflash_ready() -> bool:
return False
finally:
if connection:
connection.close()
try:
connection.close()
except Exception:
pass
def main():
@@ -35,6 +35,4 @@ class TiDBVectorTest(AbstractVectorTest):
def test_tidb_vector(setup_mock_redis, tidb_vector):
# TiDBVectorTest(vector=tidb_vector).run_all_tests()
# something wrong with tidb,ignore tidb test
return
TiDBVectorTest(vector=tidb_vector).run_all_tests()
@@ -5,10 +5,12 @@ import pytest
from faker import Faker
from core.app.entities.app_invoke_entities import InvokeFrom
from enums.cloud_plan import CloudPlan
from models.model import EndUser
from models.workflow import Workflow
from services.app_generate_service import AppGenerateService
from services.errors.app import WorkflowIdFormatError, WorkflowNotFoundError
from services.errors.llm import InvokeRateLimitError
class TestAppGenerateService:
@@ -18,9 +20,10 @@ class TestAppGenerateService:
def mock_external_service_dependencies(self):
"""Mock setup for external service dependencies."""
with (
patch("services.billing_service.BillingService") as mock_billing_service,
patch("services.app_generate_service.BillingService") as mock_billing_service,
patch("services.app_generate_service.WorkflowService") as mock_workflow_service,
patch("services.app_generate_service.RateLimit") as mock_rate_limit,
patch("services.app_generate_service.RateLimiter") as mock_rate_limiter,
patch("services.app_generate_service.CompletionAppGenerator") as mock_completion_generator,
patch("services.app_generate_service.ChatAppGenerator") as mock_chat_generator,
patch("services.app_generate_service.AgentChatAppGenerator") as mock_agent_chat_generator,
@@ -28,13 +31,9 @@ class TestAppGenerateService:
patch("services.app_generate_service.WorkflowAppGenerator") as mock_workflow_generator,
patch("services.account_service.FeatureService") as mock_account_feature_service,
patch("services.app_generate_service.dify_config") as mock_dify_config,
patch("configs.dify_config") as mock_global_dify_config,
):
# Setup default mock returns for billing service
mock_billing_service.update_tenant_feature_plan_usage.return_value = {
"result": "success",
"history_id": "test_history_id",
}
mock_billing_service.get_info.return_value = {"subscription": {"plan": CloudPlan.SANDBOX}}
# Setup default mock returns for workflow service
mock_workflow_service_instance = mock_workflow_service.return_value
@@ -48,6 +47,10 @@ class TestAppGenerateService:
mock_rate_limit_instance.generate.return_value = ["test_response"]
mock_rate_limit_instance.exit.return_value = None
mock_rate_limiter_instance = mock_rate_limiter.return_value
mock_rate_limiter_instance.is_rate_limited.return_value = False
mock_rate_limiter_instance.increment_rate_limit.return_value = None
# Setup default mock returns for app generators
mock_completion_generator_instance = mock_completion_generator.return_value
mock_completion_generator_instance.generate.return_value = ["completion_response"]
@@ -84,14 +87,11 @@ class TestAppGenerateService:
mock_dify_config.APP_MAX_ACTIVE_REQUESTS = 100
mock_dify_config.APP_DAILY_RATE_LIMIT = 1000
mock_global_dify_config.BILLING_ENABLED = False
mock_global_dify_config.APP_MAX_ACTIVE_REQUESTS = 100
mock_global_dify_config.APP_DAILY_RATE_LIMIT = 1000
yield {
"billing_service": mock_billing_service,
"workflow_service": mock_workflow_service,
"rate_limit": mock_rate_limit,
"rate_limiter": mock_rate_limiter,
"completion_generator": mock_completion_generator,
"chat_generator": mock_chat_generator,
"agent_chat_generator": mock_agent_chat_generator,
@@ -99,7 +99,6 @@ class TestAppGenerateService:
"workflow_generator": mock_workflow_generator,
"account_feature_service": mock_account_feature_service,
"dify_config": mock_dify_config,
"global_dify_config": mock_global_dify_config,
}
def _create_test_app_and_account(self, db_session_with_containers, mock_external_service_dependencies, mode="chat"):
@@ -430,9 +429,13 @@ class TestAppGenerateService:
db_session_with_containers, mock_external_service_dependencies, mode="completion"
)
# Setup billing service mock for sandbox plan
mock_external_service_dependencies["billing_service"].get_info.return_value = {
"subscription": {"plan": CloudPlan.SANDBOX}
}
# Set BILLING_ENABLED to True for this test
mock_external_service_dependencies["dify_config"].BILLING_ENABLED = True
mock_external_service_dependencies["global_dify_config"].BILLING_ENABLED = True
# Setup test arguments
args = {"inputs": {"query": fake.text(max_nb_chars=50)}, "response_mode": "streaming"}
@@ -445,8 +448,41 @@ class TestAppGenerateService:
# Verify the result
assert result == ["test_response"]
# Verify billing service was called to consume quota
mock_external_service_dependencies["billing_service"].update_tenant_feature_plan_usage.assert_called_once()
# Verify billing service was called
mock_external_service_dependencies["billing_service"].get_info.assert_called_once_with(app.tenant_id)
def test_generate_with_rate_limit_exceeded(self, db_session_with_containers, mock_external_service_dependencies):
"""
Test generation when rate limit is exceeded.
"""
fake = Faker()
app, account = self._create_test_app_and_account(
db_session_with_containers, mock_external_service_dependencies, mode="completion"
)
# Setup billing service mock for sandbox plan
mock_external_service_dependencies["billing_service"].get_info.return_value = {
"subscription": {"plan": CloudPlan.SANDBOX}
}
# Set BILLING_ENABLED to True for this test
mock_external_service_dependencies["dify_config"].BILLING_ENABLED = True
# Setup system rate limiter to return rate limited
with patch("services.app_generate_service.AppGenerateService.system_rate_limiter") as mock_system_rate_limiter:
mock_system_rate_limiter.is_rate_limited.return_value = True
# Setup test arguments
args = {"inputs": {"query": fake.text(max_nb_chars=50)}, "response_mode": "streaming"}
# Execute the method under test and expect rate limit error
with pytest.raises(InvokeRateLimitError) as exc_info:
AppGenerateService.generate(
app_model=app, user=account, args=args, invoke_from=InvokeFrom.SERVICE_API, streaming=True
)
# Verify error message
assert "Rate limit exceeded" in str(exc_info.value)
def test_generate_with_invalid_app_mode(self, db_session_with_containers, mock_external_service_dependencies):
"""
@@ -23,13 +23,11 @@ from core.mcp.auth.auth_flow import (
)
from core.mcp.entities import AuthActionType, AuthResult
from core.mcp.types import (
LATEST_PROTOCOL_VERSION,
OAuthClientInformation,
OAuthClientInformationFull,
OAuthClientMetadata,
OAuthMetadata,
OAuthTokens,
ProtectedResourceMetadata,
)
@@ -156,7 +154,7 @@ class TestOAuthDiscovery:
assert auth_url == "https://auth.example.com"
mock_get.assert_called_once_with(
"https://api.example.com/.well-known/oauth-protected-resource",
headers={"MCP-Protocol-Version": LATEST_PROTOCOL_VERSION, "User-Agent": "Dify"},
headers={"MCP-Protocol-Version": "2025-03-26", "User-Agent": "Dify"},
)
@patch("core.helper.ssrf_proxy.get")
@@ -185,61 +183,59 @@ class TestOAuthDiscovery:
assert auth_url == "https://auth.example.com"
mock_get.assert_called_once_with(
"https://api.example.com/.well-known/oauth-protected-resource?query=1#fragment",
headers={"MCP-Protocol-Version": LATEST_PROTOCOL_VERSION, "User-Agent": "Dify"},
headers={"MCP-Protocol-Version": "2025-03-26", "User-Agent": "Dify"},
)
def test_discover_oauth_metadata_with_resource_discovery(self):
@patch("core.helper.ssrf_proxy.get")
def test_discover_oauth_metadata_with_resource_discovery(self, mock_get):
"""Test OAuth metadata discovery with resource discovery support."""
with patch("core.mcp.auth.auth_flow.discover_protected_resource_metadata") as mock_prm:
with patch("core.mcp.auth.auth_flow.discover_oauth_authorization_server_metadata") as mock_asm:
# Mock protected resource metadata with auth server URL
mock_prm.return_value = ProtectedResourceMetadata(
resource="https://api.example.com",
authorization_servers=["https://auth.example.com"],
)
with patch("core.mcp.auth.auth_flow.check_support_resource_discovery") as mock_check:
mock_check.return_value = (True, "https://auth.example.com")
# Mock OAuth authorization server metadata
mock_asm.return_value = OAuthMetadata(
authorization_endpoint="https://auth.example.com/authorize",
token_endpoint="https://auth.example.com/token",
response_types_supported=["code"],
)
mock_response = Mock()
mock_response.status_code = 200
mock_response.is_success = True
mock_response.json.return_value = {
"authorization_endpoint": "https://auth.example.com/authorize",
"token_endpoint": "https://auth.example.com/token",
"response_types_supported": ["code"],
}
mock_get.return_value = mock_response
oauth_metadata, prm, scope = discover_oauth_metadata("https://api.example.com")
metadata = discover_oauth_metadata("https://api.example.com")
assert oauth_metadata is not None
assert oauth_metadata.authorization_endpoint == "https://auth.example.com/authorize"
assert oauth_metadata.token_endpoint == "https://auth.example.com/token"
assert prm is not None
assert prm.authorization_servers == ["https://auth.example.com"]
assert metadata is not None
assert metadata.authorization_endpoint == "https://auth.example.com/authorize"
assert metadata.token_endpoint == "https://auth.example.com/token"
mock_get.assert_called_once_with(
"https://auth.example.com/.well-known/oauth-authorization-server",
headers={"MCP-Protocol-Version": "2025-03-26"},
)
# Verify the discovery functions were called
mock_prm.assert_called_once()
mock_asm.assert_called_once()
def test_discover_oauth_metadata_without_resource_discovery(self):
@patch("core.helper.ssrf_proxy.get")
def test_discover_oauth_metadata_without_resource_discovery(self, mock_get):
"""Test OAuth metadata discovery without resource discovery."""
with patch("core.mcp.auth.auth_flow.discover_protected_resource_metadata") as mock_prm:
with patch("core.mcp.auth.auth_flow.discover_oauth_authorization_server_metadata") as mock_asm:
# Mock no protected resource metadata
mock_prm.return_value = None
with patch("core.mcp.auth.auth_flow.check_support_resource_discovery") as mock_check:
mock_check.return_value = (False, "")
# Mock OAuth authorization server metadata
mock_asm.return_value = OAuthMetadata(
authorization_endpoint="https://api.example.com/oauth/authorize",
token_endpoint="https://api.example.com/oauth/token",
response_types_supported=["code"],
)
mock_response = Mock()
mock_response.status_code = 200
mock_response.is_success = True
mock_response.json.return_value = {
"authorization_endpoint": "https://api.example.com/oauth/authorize",
"token_endpoint": "https://api.example.com/oauth/token",
"response_types_supported": ["code"],
}
mock_get.return_value = mock_response
oauth_metadata, prm, scope = discover_oauth_metadata("https://api.example.com")
metadata = discover_oauth_metadata("https://api.example.com")
assert oauth_metadata is not None
assert oauth_metadata.authorization_endpoint == "https://api.example.com/oauth/authorize"
assert prm is None
# Verify the discovery functions were called
mock_prm.assert_called_once()
mock_asm.assert_called_once()
assert metadata is not None
assert metadata.authorization_endpoint == "https://api.example.com/oauth/authorize"
mock_get.assert_called_once_with(
"https://api.example.com/.well-known/oauth-authorization-server",
headers={"MCP-Protocol-Version": "2025-03-26"},
)
@patch("core.helper.ssrf_proxy.get")
def test_discover_oauth_metadata_not_found(self, mock_get):
@@ -251,9 +247,9 @@ class TestOAuthDiscovery:
mock_response.status_code = 404
mock_get.return_value = mock_response
oauth_metadata, prm, scope = discover_oauth_metadata("https://api.example.com")
metadata = discover_oauth_metadata("https://api.example.com")
assert oauth_metadata is None
assert metadata is None
class TestAuthorizationFlow:
@@ -346,7 +342,6 @@ class TestAuthorizationFlow:
"""Test successful authorization code exchange."""
mock_response = Mock()
mock_response.is_success = True
mock_response.headers = {"content-type": "application/json"}
mock_response.json.return_value = {
"access_token": "new-access-token",
"token_type": "Bearer",
@@ -417,7 +412,6 @@ class TestAuthorizationFlow:
"""Test successful token refresh."""
mock_response = Mock()
mock_response.is_success = True
mock_response.headers = {"content-type": "application/json"}
mock_response.json.return_value = {
"access_token": "refreshed-access-token",
"token_type": "Bearer",
@@ -583,15 +577,11 @@ class TestAuthOrchestration:
def test_auth_new_registration(self, mock_start_auth, mock_register, mock_discover, mock_provider, mock_service):
"""Test auth flow for new client registration."""
# Setup
mock_discover.return_value = (
OAuthMetadata(
authorization_endpoint="https://auth.example.com/authorize",
token_endpoint="https://auth.example.com/token",
response_types_supported=["code"],
grant_types_supported=["authorization_code"],
),
None,
None,
mock_discover.return_value = OAuthMetadata(
authorization_endpoint="https://auth.example.com/authorize",
token_endpoint="https://auth.example.com/token",
response_types_supported=["code"],
grant_types_supported=["authorization_code"],
)
mock_register.return_value = OAuthClientInformationFull(
client_id="new-client-id",
@@ -629,15 +619,11 @@ class TestAuthOrchestration:
def test_auth_exchange_code(self, mock_exchange, mock_retrieve_state, mock_discover, mock_provider, mock_service):
"""Test auth flow for exchanging authorization code."""
# Setup metadata discovery
mock_discover.return_value = (
OAuthMetadata(
authorization_endpoint="https://auth.example.com/authorize",
token_endpoint="https://auth.example.com/token",
response_types_supported=["code"],
grant_types_supported=["authorization_code"],
),
None,
None,
mock_discover.return_value = OAuthMetadata(
authorization_endpoint="https://auth.example.com/authorize",
token_endpoint="https://auth.example.com/token",
response_types_supported=["code"],
grant_types_supported=["authorization_code"],
)
# Setup existing client
@@ -676,15 +662,11 @@ class TestAuthOrchestration:
def test_auth_exchange_code_without_state(self, mock_discover, mock_provider, mock_service):
"""Test auth flow fails when exchanging code without state."""
# Setup metadata discovery
mock_discover.return_value = (
OAuthMetadata(
authorization_endpoint="https://auth.example.com/authorize",
token_endpoint="https://auth.example.com/token",
response_types_supported=["code"],
grant_types_supported=["authorization_code"],
),
None,
None,
mock_discover.return_value = OAuthMetadata(
authorization_endpoint="https://auth.example.com/authorize",
token_endpoint="https://auth.example.com/token",
response_types_supported=["code"],
grant_types_supported=["authorization_code"],
)
mock_provider.retrieve_client_information.return_value = OAuthClientInformation(client_id="existing-client")
@@ -716,15 +698,11 @@ class TestAuthOrchestration:
mock_refresh.return_value = new_tokens
with patch("core.mcp.auth.auth_flow.discover_oauth_metadata") as mock_discover:
mock_discover.return_value = (
OAuthMetadata(
authorization_endpoint="https://auth.example.com/authorize",
token_endpoint="https://auth.example.com/token",
response_types_supported=["code"],
grant_types_supported=["authorization_code"],
),
None,
None,
mock_discover.return_value = OAuthMetadata(
authorization_endpoint="https://auth.example.com/authorize",
token_endpoint="https://auth.example.com/token",
response_types_supported=["code"],
grant_types_supported=["authorization_code"],
)
result = auth(mock_provider)
@@ -747,15 +725,11 @@ class TestAuthOrchestration:
def test_auth_registration_fails_with_code(self, mock_discover, mock_provider, mock_service):
"""Test auth fails when no client info exists but code is provided."""
# Setup metadata discovery
mock_discover.return_value = (
OAuthMetadata(
authorization_endpoint="https://auth.example.com/authorize",
token_endpoint="https://auth.example.com/token",
response_types_supported=["code"],
grant_types_supported=["authorization_code"],
),
None,
None,
mock_discover.return_value = OAuthMetadata(
authorization_endpoint="https://auth.example.com/authorize",
token_endpoint="https://auth.example.com/token",
response_types_supported=["code"],
grant_types_supported=["authorization_code"],
)
mock_provider.retrieve_client_information.return_value = None
@@ -139,9 +139,7 @@ def test_sse_client_error_handling():
with patch("core.mcp.client.sse_client.create_ssrf_proxy_mcp_http_client") as mock_client_factory:
with patch("core.mcp.client.sse_client.ssrf_proxy_sse_connect") as mock_sse_connect:
# Mock 401 HTTP error
mock_response = Mock(status_code=401)
mock_response.headers = {"WWW-Authenticate": 'Bearer realm="example"'}
mock_error = httpx.HTTPStatusError("Unauthorized", request=Mock(), response=mock_response)
mock_error = httpx.HTTPStatusError("Unauthorized", request=Mock(), response=Mock(status_code=401))
mock_sse_connect.side_effect = mock_error
with pytest.raises(MCPAuthError):
@@ -152,9 +150,7 @@ def test_sse_client_error_handling():
with patch("core.mcp.client.sse_client.create_ssrf_proxy_mcp_http_client") as mock_client_factory:
with patch("core.mcp.client.sse_client.ssrf_proxy_sse_connect") as mock_sse_connect:
# Mock other HTTP error
mock_response = Mock(status_code=500)
mock_response.headers = {}
mock_error = httpx.HTTPStatusError("Server Error", request=Mock(), response=mock_response)
mock_error = httpx.HTTPStatusError("Server Error", request=Mock(), response=Mock(status_code=500))
mock_sse_connect.side_effect = mock_error
with pytest.raises(MCPConnectionError):
+1 -1
View File
@@ -58,7 +58,7 @@ class TestConstants:
def test_protocol_versions(self):
"""Test protocol version constants."""
assert LATEST_PROTOCOL_VERSION == "2025-06-18"
assert LATEST_PROTOCOL_VERSION == "2025-03-26"
assert SERVER_LATEST_PROTOCOL_VERSION == "2024-11-05"
def test_error_codes(self):
@@ -1,189 +0,0 @@
"""Tests for dispatcher command checking behavior."""
from __future__ import annotations
import queue
from datetime import datetime
from unittest import mock
from core.workflow.entities.pause_reason import SchedulingPause
from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
from core.workflow.graph_engine.event_management.event_handlers import EventHandler
from core.workflow.graph_engine.orchestration.dispatcher import Dispatcher
from core.workflow.graph_engine.orchestration.execution_coordinator import ExecutionCoordinator
from core.workflow.graph_events import (
GraphNodeEventBase,
NodeRunPauseRequestedEvent,
NodeRunStartedEvent,
NodeRunSucceededEvent,
)
from core.workflow.node_events import NodeRunResult
def test_dispatcher_should_consume_remains_events_after_pause():
event_queue = queue.Queue()
event_queue.put(
GraphNodeEventBase(
id="test",
node_id="test",
node_type=NodeType.START,
)
)
event_handler = mock.Mock(spec=EventHandler)
execution_coordinator = mock.Mock(spec=ExecutionCoordinator)
execution_coordinator.paused.return_value = True
dispatcher = Dispatcher(
event_queue=event_queue,
event_handler=event_handler,
execution_coordinator=execution_coordinator,
)
dispatcher._dispatcher_loop()
assert event_queue.empty()
class _StubExecutionCoordinator:
"""Stub execution coordinator that tracks command checks."""
def __init__(self) -> None:
self.command_checks = 0
self.scaling_checks = 0
self.execution_complete = False
self.failed = False
self._paused = False
def process_commands(self) -> None:
self.command_checks += 1
def check_scaling(self) -> None:
self.scaling_checks += 1
@property
def paused(self) -> bool:
return self._paused
@property
def aborted(self) -> bool:
return False
def mark_complete(self) -> None:
self.execution_complete = True
def mark_failed(self, error: Exception) -> None: # pragma: no cover - defensive, not triggered in tests
self.failed = True
class _StubEventHandler:
"""Minimal event handler that marks execution complete after handling an event."""
def __init__(self, coordinator: _StubExecutionCoordinator) -> None:
self._coordinator = coordinator
self.events = []
def dispatch(self, event) -> None:
self.events.append(event)
self._coordinator.mark_complete()
def _run_dispatcher_for_event(event) -> int:
"""Run the dispatcher loop for a single event and return command check count."""
event_queue: queue.Queue = queue.Queue()
event_queue.put(event)
coordinator = _StubExecutionCoordinator()
event_handler = _StubEventHandler(coordinator)
dispatcher = Dispatcher(
event_queue=event_queue,
event_handler=event_handler,
execution_coordinator=coordinator,
)
dispatcher._dispatcher_loop()
return coordinator.command_checks
def _make_started_event() -> NodeRunStartedEvent:
return NodeRunStartedEvent(
id="start-event",
node_id="node-1",
node_type=NodeType.CODE,
node_title="Test Node",
start_at=datetime.utcnow(),
)
def _make_succeeded_event() -> NodeRunSucceededEvent:
return NodeRunSucceededEvent(
id="success-event",
node_id="node-1",
node_type=NodeType.CODE,
node_title="Test Node",
start_at=datetime.utcnow(),
node_run_result=NodeRunResult(status=WorkflowNodeExecutionStatus.SUCCEEDED),
)
def test_dispatcher_checks_commands_during_idle_and_on_completion() -> None:
"""Dispatcher polls commands when idle and after completion events."""
started_checks = _run_dispatcher_for_event(_make_started_event())
succeeded_checks = _run_dispatcher_for_event(_make_succeeded_event())
assert started_checks == 2
assert succeeded_checks == 3
class _PauseStubEventHandler:
"""Minimal event handler that marks execution complete after handling an event."""
def __init__(self, coordinator: _StubExecutionCoordinator) -> None:
self._coordinator = coordinator
self.events = []
def dispatch(self, event) -> None:
self.events.append(event)
if isinstance(event, NodeRunPauseRequestedEvent):
self._coordinator.mark_complete()
def test_dispatcher_drain_event_queue():
events = [
NodeRunStartedEvent(
id="start-event",
node_id="node-1",
node_type=NodeType.CODE,
node_title="Code",
start_at=datetime.utcnow(),
),
NodeRunPauseRequestedEvent(
id="pause-event",
node_id="node-1",
node_type=NodeType.CODE,
reason=SchedulingPause(message="test pause"),
),
NodeRunSucceededEvent(
id="success-event",
node_id="node-1",
node_type=NodeType.CODE,
start_at=datetime.utcnow(),
node_run_result=NodeRunResult(status=WorkflowNodeExecutionStatus.SUCCEEDED),
),
]
event_queue: queue.Queue = queue.Queue()
for e in events:
event_queue.put(e)
coordinator = _StubExecutionCoordinator()
event_handler = _PauseStubEventHandler(coordinator)
dispatcher = Dispatcher(
event_queue=event_queue,
event_handler=event_handler,
execution_coordinator=coordinator,
)
dispatcher._dispatcher_loop()
# ensure all events are drained.
assert event_queue.empty()
@@ -3,17 +3,13 @@
import time
from unittest.mock import MagicMock
from core.app.entities.app_invoke_entities import InvokeFrom
from core.workflow.entities.graph_init_params import GraphInitParams
from core.workflow.entities.pause_reason import SchedulingPause
from core.workflow.graph import Graph
from core.workflow.graph_engine import GraphEngine
from core.workflow.graph_engine.command_channels import InMemoryChannel
from core.workflow.graph_engine.entities.commands import AbortCommand, CommandType, PauseCommand
from core.workflow.graph_events import GraphRunAbortedEvent, GraphRunPausedEvent, GraphRunStartedEvent
from core.workflow.nodes.start.start_node import StartNode
from core.workflow.runtime import GraphRuntimeState, VariablePool
from models.enums import UserFrom
def test_abort_command():
@@ -30,23 +26,11 @@ def test_abort_command():
mock_graph.root_node.id = "start"
# Create mock nodes with required attributes - using shared runtime state
start_node = StartNode(
id="start",
config={"id": "start"},
graph_init_params=GraphInitParams(
tenant_id="test_tenant",
app_id="test_app",
workflow_id="test_workflow",
graph_config={},
user_id="test_user",
user_from=UserFrom.ACCOUNT,
invoke_from=InvokeFrom.DEBUGGER,
call_depth=0,
),
graph_runtime_state=shared_runtime_state,
)
start_node.init_node_data({"title": "start", "variables": []})
mock_graph.nodes["start"] = start_node
mock_start_node = MagicMock()
mock_start_node.state = None
mock_start_node.id = "start"
mock_start_node.graph_runtime_state = shared_runtime_state # Use shared instance
mock_graph.nodes["start"] = mock_start_node
# Mock graph methods
mock_graph.get_outgoing_edges = MagicMock(return_value=[])
@@ -140,23 +124,11 @@ def test_pause_command():
mock_graph.root_node = MagicMock()
mock_graph.root_node.id = "start"
start_node = StartNode(
id="start",
config={"id": "start"},
graph_init_params=GraphInitParams(
tenant_id="test_tenant",
app_id="test_app",
workflow_id="test_workflow",
graph_config={},
user_id="test_user",
user_from=UserFrom.ACCOUNT,
invoke_from=InvokeFrom.DEBUGGER,
call_depth=0,
),
graph_runtime_state=shared_runtime_state,
)
start_node.init_node_data({"title": "start", "variables": []})
mock_graph.nodes["start"] = start_node
mock_start_node = MagicMock()
mock_start_node.state = None
mock_start_node.id = "start"
mock_start_node.graph_runtime_state = shared_runtime_state
mock_graph.nodes["start"] = mock_start_node
mock_graph.get_outgoing_edges = MagicMock(return_value=[])
mock_graph.get_incoming_edges = MagicMock(return_value=[])
@@ -181,5 +153,5 @@ def test_pause_command():
assert pause_events[0].reason == SchedulingPause(message="User requested pause")
graph_execution = engine.graph_runtime_state.graph_execution
assert graph_execution.paused
assert graph_execution.is_paused
assert graph_execution.pause_reason == SchedulingPause(message="User requested pause")
@@ -0,0 +1,109 @@
"""Tests for dispatcher command checking behavior."""
from __future__ import annotations
import queue
from datetime import datetime
from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
from core.workflow.graph_engine.event_management.event_manager import EventManager
from core.workflow.graph_engine.orchestration.dispatcher import Dispatcher
from core.workflow.graph_events import NodeRunStartedEvent, NodeRunSucceededEvent
from core.workflow.node_events import NodeRunResult
class _StubExecutionCoordinator:
"""Stub execution coordinator that tracks command checks."""
def __init__(self) -> None:
self.command_checks = 0
self.scaling_checks = 0
self._execution_complete = False
self.mark_complete_called = False
self.failed = False
self._paused = False
def check_commands(self) -> None:
self.command_checks += 1
def check_scaling(self) -> None:
self.scaling_checks += 1
@property
def is_paused(self) -> bool:
return self._paused
def is_execution_complete(self) -> bool:
return self._execution_complete
def mark_complete(self) -> None:
self.mark_complete_called = True
def mark_failed(self, error: Exception) -> None: # pragma: no cover - defensive, not triggered in tests
self.failed = True
def set_execution_complete(self) -> None:
self._execution_complete = True
class _StubEventHandler:
"""Minimal event handler that marks execution complete after handling an event."""
def __init__(self, coordinator: _StubExecutionCoordinator) -> None:
self._coordinator = coordinator
self.events = []
def dispatch(self, event) -> None:
self.events.append(event)
self._coordinator.set_execution_complete()
def _run_dispatcher_for_event(event) -> int:
"""Run the dispatcher loop for a single event and return command check count."""
event_queue: queue.Queue = queue.Queue()
event_queue.put(event)
coordinator = _StubExecutionCoordinator()
event_handler = _StubEventHandler(coordinator)
event_manager = EventManager()
dispatcher = Dispatcher(
event_queue=event_queue,
event_handler=event_handler,
event_collector=event_manager,
execution_coordinator=coordinator,
)
dispatcher._dispatcher_loop()
return coordinator.command_checks
def _make_started_event() -> NodeRunStartedEvent:
return NodeRunStartedEvent(
id="start-event",
node_id="node-1",
node_type=NodeType.CODE,
node_title="Test Node",
start_at=datetime.utcnow(),
)
def _make_succeeded_event() -> NodeRunSucceededEvent:
return NodeRunSucceededEvent(
id="success-event",
node_id="node-1",
node_type=NodeType.CODE,
node_title="Test Node",
start_at=datetime.utcnow(),
node_run_result=NodeRunResult(status=WorkflowNodeExecutionStatus.SUCCEEDED),
)
def test_dispatcher_checks_commands_during_idle_and_on_completion() -> None:
"""Dispatcher polls commands when idle and after completion events."""
started_checks = _run_dispatcher_for_event(_make_started_event())
succeeded_checks = _run_dispatcher_for_event(_make_succeeded_event())
assert started_checks == 1
assert succeeded_checks == 2
@@ -48,3 +48,15 @@ def test_handle_pause_noop_when_execution_running() -> None:
worker_pool.stop.assert_not_called()
state_manager.clear_executing.assert_not_called()
def test_is_execution_complete_when_paused() -> None:
"""Paused execution should be treated as complete."""
graph_execution = GraphExecution(workflow_id="workflow")
graph_execution.start()
graph_execution.pause("Awaiting input")
coordinator, state_manager, _worker_pool = _build_coordinator(graph_execution)
state_manager.is_execution_complete.return_value = False
assert coordinator.is_execution_complete()
@@ -1,33 +0,0 @@
import sqlalchemy as sa
from models.dataset import Document
from services.dataset_service import DocumentService
def test_normalize_display_status_alias_mapping():
assert DocumentService.normalize_display_status("ACTIVE") == "available"
assert DocumentService.normalize_display_status("enabled") == "available"
assert DocumentService.normalize_display_status("archived") == "archived"
assert DocumentService.normalize_display_status("unknown") is None
def test_build_display_status_filters_available():
filters = DocumentService.build_display_status_filters("available")
assert len(filters) == 3
for condition in filters:
assert condition is not None
def test_apply_display_status_filter_applies_when_status_present():
query = sa.select(Document)
filtered = DocumentService.apply_display_status_filter(query, "queuing")
compiled = str(filtered.compile(compile_kwargs={"literal_binds": True}))
assert "WHERE" in compiled
assert "document.indexing_status = 'waiting'" in compiled
def test_apply_display_status_filter_returns_same_when_invalid():
query = sa.select(Document)
filtered = DocumentService.apply_display_status_filter(query, "invalid")
compiled = str(filtered.compile(compile_kwargs={"literal_binds": True}))
assert "WHERE" not in compiled
@@ -1,18 +0,0 @@
from core.app.apps.workflow.app_generator import SKIP_PREPARE_USER_INPUTS_KEY
from services.workflow.entities import WebhookTriggerData
from tasks import async_workflow_tasks
def test_build_generator_args_sets_skip_flag_for_webhook():
trigger_data = WebhookTriggerData(
app_id="app",
tenant_id="tenant",
workflow_id="workflow",
root_node_id="node",
inputs={"webhook_data": {"body": {"foo": "bar"}}},
)
args = async_workflow_tasks._build_generator_args(trigger_data)
assert args[SKIP_PREPARE_USER_INPUTS_KEY] is True
assert args["inputs"]["webhook_data"]["body"]["foo"] == "bar"
Generated
+803 -795
View File
File diff suppressed because it is too large Load Diff
-20
View File
@@ -11,7 +11,6 @@ show_help() {
echo " -c, --concurrency NUM Number of worker processes (default: 1)"
echo " -P, --pool POOL Pool implementation (default: gevent)"
echo " --loglevel LEVEL Log level (default: INFO)"
echo " -e, --env-file FILE Path to an env file to source before starting"
echo " -h, --help Show this help message"
echo ""
echo "Examples:"
@@ -45,8 +44,6 @@ CONCURRENCY=1
POOL="gevent"
LOGLEVEL="INFO"
ENV_FILE=""
while [[ $# -gt 0 ]]; do
case $1 in
-q|--queues)
@@ -65,10 +62,6 @@ while [[ $# -gt 0 ]]; do
LOGLEVEL="$2"
shift 2
;;
-e|--env-file)
ENV_FILE="$2"
shift 2
;;
-h|--help)
show_help
exit 0
@@ -84,19 +77,6 @@ done
SCRIPT_DIR="$(dirname "$(realpath "$0")")"
cd "$SCRIPT_DIR/.."
if [[ -n "${ENV_FILE}" ]]; then
if [[ ! -f "${ENV_FILE}" ]]; then
echo "Env file ${ENV_FILE} not found"
exit 1
fi
echo "Loading environment variables from ${ENV_FILE}"
# Export everything sourced from the env file
set -a
source "${ENV_FILE}"
set +a
fi
# If no queues specified, use edition-based defaults
if [[ -z "${QUEUES}" ]]; then
# Get EDITION from environment, default to SELF_HOSTED (community edition)
+5 -4
View File
@@ -224,7 +224,7 @@ NEXT_PUBLIC_ENABLE_SINGLE_DOLLAR_LATEX=false
# ------------------------------
# Database Configuration
# The database uses PostgreSQL or MySQL. OceanBase and seekdb are also supported. Please use the public schema.
# The database uses PostgreSQL or MySQL. OceanBase and SeekDB are also supported. Please use the public schema.
# It is consistent with the configuration in the database service below.
# You can adjust the database configuration according to your needs.
# ------------------------------
@@ -393,9 +393,10 @@ WEB_API_CORS_ALLOW_ORIGINS=*
# Specifies the allowed origins for cross-origin requests to the console API,
# e.g. https://cloud.dify.ai or * for all origins.
CONSOLE_CORS_ALLOW_ORIGINS=*
# When the frontend and backend run on different subdomains, set COOKIE_DOMAIN to the sites top-level domain (e.g., `example.com`). Leading dots are optional.
# Set COOKIE_DOMAIN when the console frontend and API are on different subdomains.
# Provide the registrable domain (e.g. example.com); leading dots are optional.
COOKIE_DOMAIN=
# When the frontend and backend run on different subdomains, set NEXT_PUBLIC_COOKIE_DOMAIN=1.
# The frontend reads NEXT_PUBLIC_COOKIE_DOMAIN to align cookie handling with the API.
NEXT_PUBLIC_COOKIE_DOMAIN=
# ------------------------------
@@ -529,7 +530,7 @@ WEAVIATE_GRPC_ENDPOINT=grpc://weaviate:50051
# For OceanBase metadata database configuration, available when `DB_TYPE` is `mysql` and `COMPOSE_PROFILES` includes `oceanbase`.
# For OceanBase vector database configuration, available when `VECTOR_STORE` is `oceanbase`
# If you want to use OceanBase as both vector database and metadata database, you need to set `DB_TYPE` to `mysql`, `COMPOSE_PROFILES` is `oceanbase`, and set Database Configuration is the same as the vector database.
# seekdb is the lite version of OceanBase and shares the connection configuration with OceanBase.
# SeekDB is the lite version of OceanBase and shares the connection configuration with OceanBase.
OCEANBASE_VECTOR_HOST=oceanbase
OCEANBASE_VECTOR_PORT=2881
OCEANBASE_VECTOR_USER=root@test
+6 -6
View File
@@ -2,7 +2,7 @@ x-shared-env: &shared-api-worker-env
services:
# API service
api:
image: langgenius/dify-api:1.10.0
image: langgenius/dify-api:1.10.0-rc1
restart: always
environment:
# Use the shared environment variables.
@@ -41,7 +41,7 @@ services:
# worker service
# The Celery worker for processing all queues (dataset, workflow, mail, etc.)
worker:
image: langgenius/dify-api:1.10.0
image: langgenius/dify-api:1.10.0-rc1
restart: always
environment:
# Use the shared environment variables.
@@ -78,7 +78,7 @@ services:
# worker_beat service
# Celery beat for scheduling periodic tasks.
worker_beat:
image: langgenius/dify-api:1.10.0
image: langgenius/dify-api:1.10.0-rc1
restart: always
environment:
# Use the shared environment variables.
@@ -106,7 +106,7 @@ services:
# Frontend web application.
web:
image: langgenius/dify-web:1.10.0
image: langgenius/dify-web:1.10.0-rc1
restart: always
environment:
CONSOLE_API_URL: ${CONSOLE_API_URL:-}
@@ -243,7 +243,7 @@ services:
# plugin daemon
plugin_daemon:
image: langgenius/dify-plugin-daemon:0.4.1-local
image: langgenius/dify-plugin-daemon:0.4.0-local
restart: always
environment:
# Use the shared environment variables.
@@ -458,7 +458,7 @@ services:
start_period: 30s
timeout: 10s
# seekdb vector database
# SeekDB vector database
seekdb:
image: oceanbase/seekdb:latest
container_name: seekdb
+6 -6
View File
@@ -635,7 +635,7 @@ x-shared-env: &shared-api-worker-env
services:
# API service
api:
image: langgenius/dify-api:1.10.0
image: langgenius/dify-api:1.10.0-rc1
restart: always
environment:
# Use the shared environment variables.
@@ -674,7 +674,7 @@ services:
# worker service
# The Celery worker for processing all queues (dataset, workflow, mail, etc.)
worker:
image: langgenius/dify-api:1.10.0
image: langgenius/dify-api:1.10.0-rc1
restart: always
environment:
# Use the shared environment variables.
@@ -711,7 +711,7 @@ services:
# worker_beat service
# Celery beat for scheduling periodic tasks.
worker_beat:
image: langgenius/dify-api:1.10.0
image: langgenius/dify-api:1.10.0-rc1
restart: always
environment:
# Use the shared environment variables.
@@ -739,7 +739,7 @@ services:
# Frontend web application.
web:
image: langgenius/dify-web:1.10.0
image: langgenius/dify-web:1.10.0-rc1
restart: always
environment:
CONSOLE_API_URL: ${CONSOLE_API_URL:-}
@@ -876,7 +876,7 @@ services:
# plugin daemon
plugin_daemon:
image: langgenius/dify-plugin-daemon:0.4.1-local
image: langgenius/dify-plugin-daemon:0.4.0-local
restart: always
environment:
# Use the shared environment variables.
@@ -1091,7 +1091,7 @@ services:
start_period: 30s
timeout: 10s
# seekdb vector database
# SeekDB vector database
seekdb:
image: oceanbase/seekdb:latest
container_name: seekdb
+1 -2
View File
@@ -55,8 +55,7 @@ services:
- ./volumes/data:/data
- ./volumes/logs:/logs
command:
- server
- --config-file=/tiflash.toml
- --config=/tiflash.toml
depends_on:
- "tikv"
- "tidb"
File diff suppressed because it is too large Load Diff
@@ -1,228 +0,0 @@
"""Base client with common functionality for both sync and async clients."""
import json
import time
import logging
from typing import Dict, Callable, Optional
try:
# Python 3.10+
from typing import ParamSpec
except ImportError:
# Python < 3.10
from typing_extensions import ParamSpec
from urllib.parse import urljoin
import httpx
P = ParamSpec("P")
from .exceptions import (
DifyClientError,
APIError,
AuthenticationError,
RateLimitError,
ValidationError,
NetworkError,
TimeoutError,
)
class BaseClientMixin:
"""Mixin class providing common functionality for Dify clients."""
def __init__(
self,
api_key: str,
base_url: str = "https://api.dify.ai/v1",
timeout: float = 60.0,
max_retries: int = 3,
retry_delay: float = 1.0,
enable_logging: bool = False,
):
"""Initialize the base client.
Args:
api_key: Your Dify API key
base_url: Base URL for the Dify API
timeout: Request timeout in seconds
max_retries: Maximum number of retry attempts
retry_delay: Delay between retries in seconds
enable_logging: Enable detailed logging
"""
if not api_key:
raise ValidationError("API key is required")
self.api_key = api_key
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self.max_retries = max_retries
self.retry_delay = retry_delay
self.enable_logging = enable_logging
# Setup logging
self.logger = logging.getLogger(f"dify_client.{self.__class__.__name__.lower()}")
if enable_logging and not self.logger.handlers:
# Create console handler with formatter
handler = logging.StreamHandler()
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
handler.setFormatter(formatter)
self.logger.addHandler(handler)
self.logger.setLevel(logging.INFO)
self.enable_logging = True
else:
self.enable_logging = enable_logging
def _get_headers(self, content_type: str = "application/json") -> Dict[str, str]:
"""Get common request headers."""
return {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": content_type,
"User-Agent": "dify-client-python/0.1.12",
}
def _build_url(self, endpoint: str) -> str:
"""Build full URL from endpoint."""
return urljoin(self.base_url + "/", endpoint.lstrip("/"))
def _handle_response(self, response: httpx.Response) -> httpx.Response:
"""Handle HTTP response and raise appropriate exceptions."""
try:
if response.status_code == 401:
raise AuthenticationError(
"Authentication failed. Check your API key.",
status_code=response.status_code,
response=response.json() if response.content else None,
)
elif response.status_code == 429:
retry_after = response.headers.get("Retry-After")
raise RateLimitError(
"Rate limit exceeded. Please try again later.",
retry_after=int(retry_after) if retry_after else None,
)
elif response.status_code >= 400:
try:
error_data = response.json()
message = error_data.get("message", f"HTTP {response.status_code}")
except:
message = f"HTTP {response.status_code}: {response.text}"
raise APIError(
message,
status_code=response.status_code,
response=response.json() if response.content else None,
)
return response
except json.JSONDecodeError:
raise APIError(
f"Invalid JSON response: {response.text}",
status_code=response.status_code,
)
def _retry_request(
self,
request_func: Callable[P, httpx.Response],
request_context: str | None = None,
*args: P.args,
**kwargs: P.kwargs,
) -> httpx.Response:
"""Retry a request with exponential backoff.
Args:
request_func: Function that performs the HTTP request
request_context: Context description for logging (e.g., "GET /v1/messages")
*args: Positional arguments to pass to request_func
**kwargs: Keyword arguments to pass to request_func
Returns:
httpx.Response: Successful response
Raises:
NetworkError: On network failures after retries
TimeoutError: On timeout failures after retries
APIError: On API errors (4xx/5xx responses)
DifyClientError: On unexpected failures
"""
last_exception = None
for attempt in range(self.max_retries + 1):
try:
response = request_func(*args, **kwargs)
return response # Let caller handle response processing
except (httpx.NetworkError, httpx.TimeoutException) as e:
last_exception = e
context_msg = f" {request_context}" if request_context else ""
if attempt < self.max_retries:
delay = self.retry_delay * (2**attempt) # Exponential backoff
self.logger.warning(
f"Request failed{context_msg} (attempt {attempt + 1}/{self.max_retries + 1}): {e}. "
f"Retrying in {delay:.2f} seconds..."
)
time.sleep(delay)
else:
self.logger.error(f"Request failed{context_msg} after {self.max_retries + 1} attempts: {e}")
# Convert to custom exceptions
if isinstance(e, httpx.TimeoutException):
from .exceptions import TimeoutError
raise TimeoutError(f"Request timed out after {self.max_retries} retries{context_msg}") from e
else:
from .exceptions import NetworkError
raise NetworkError(
f"Network error after {self.max_retries} retries{context_msg}: {str(e)}"
) from e
if last_exception:
raise last_exception
raise DifyClientError("Request failed after retries")
def _validate_params(self, **params) -> None:
"""Validate request parameters."""
for key, value in params.items():
if value is None:
continue
# String validations
if isinstance(value, str):
if not value.strip():
raise ValidationError(f"Parameter '{key}' cannot be empty or whitespace only")
if len(value) > 10000:
raise ValidationError(f"Parameter '{key}' exceeds maximum length of 10000 characters")
# List validations
elif isinstance(value, list):
if len(value) > 1000:
raise ValidationError(f"Parameter '{key}' exceeds maximum size of 1000 items")
# Dictionary validations
elif isinstance(value, dict):
if len(value) > 100:
raise ValidationError(f"Parameter '{key}' exceeds maximum size of 100 items")
# Type-specific validations
if key == "user" and not isinstance(value, str):
raise ValidationError(f"Parameter '{key}' must be a string")
elif key in ["page", "limit", "page_size"] and not isinstance(value, int):
raise ValidationError(f"Parameter '{key}' must be an integer")
elif key == "files" and not isinstance(value, (list, dict)):
raise ValidationError(f"Parameter '{key}' must be a list or dict")
elif key == "rating" and value not in ["like", "dislike"]:
raise ValidationError(f"Parameter '{key}' must be 'like' or 'dislike'")
def _log_request(self, method: str, url: str, **kwargs) -> None:
"""Log request details."""
self.logger.info(f"Making {method} request to {url}")
if kwargs.get("json"):
self.logger.debug(f"Request body: {kwargs['json']}")
if kwargs.get("params"):
self.logger.debug(f"Query params: {kwargs['params']}")
def _log_response(self, response: httpx.Response) -> None:
"""Log response details."""
self.logger.info(f"Received response: {response.status_code} ({len(response.content)} bytes)")
+24 -396
View File
@@ -1,20 +1,11 @@
import json
import logging
import os
from typing import Literal, Dict, List, Any, IO, Optional, Union
from typing import Literal, Dict, List, Any, IO
import httpx
from .base_client import BaseClientMixin
from .exceptions import (
APIError,
AuthenticationError,
RateLimitError,
ValidationError,
FileUploadError,
)
class DifyClient(BaseClientMixin):
class DifyClient:
"""Synchronous Dify API client.
This client uses httpx.Client for efficient connection pooling and resource management.
@@ -30,9 +21,6 @@ class DifyClient(BaseClientMixin):
api_key: str,
base_url: str = "https://api.dify.ai/v1",
timeout: float = 60.0,
max_retries: int = 3,
retry_delay: float = 1.0,
enable_logging: bool = False,
):
"""Initialize the Dify client.
@@ -40,13 +28,9 @@ class DifyClient(BaseClientMixin):
api_key: Your Dify API key
base_url: Base URL for the Dify API
timeout: Request timeout in seconds (default: 60.0)
max_retries: Maximum number of retry attempts (default: 3)
retry_delay: Delay between retries in seconds (default: 1.0)
enable_logging: Whether to enable request logging (default: True)
"""
# Initialize base client functionality
BaseClientMixin.__init__(self, api_key, base_url, timeout, max_retries, retry_delay, enable_logging)
self.api_key = api_key
self.base_url = base_url
self._client = httpx.Client(
base_url=base_url,
timeout=httpx.Timeout(timeout, connect=5.0),
@@ -69,12 +53,12 @@ class DifyClient(BaseClientMixin):
self,
method: str,
endpoint: str,
json: Dict[str, Any] | None = None,
params: Dict[str, Any] | None = None,
json: dict | None = None,
params: dict | None = None,
stream: bool = False,
**kwargs,
):
"""Send an HTTP request to the Dify API with retry logic.
"""Send an HTTP request to the Dify API.
Args:
method: HTTP method (GET, POST, PUT, PATCH, DELETE)
@@ -87,91 +71,23 @@ class DifyClient(BaseClientMixin):
Returns:
httpx.Response object
"""
# Validate parameters
if json:
self._validate_params(**json)
if params:
self._validate_params(**params)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
def make_request():
"""Inner function to perform the actual HTTP request."""
# Log request if logging is enabled
if self.enable_logging:
self.logger.info(f"Sending {method} request to {endpoint}")
# Debug logging for detailed information
if self.logger.isEnabledFor(logging.DEBUG):
if json:
self.logger.debug(f"Request body: {json}")
if params:
self.logger.debug(f"Request params: {params}")
# httpx.Client automatically prepends base_url
response = self._client.request(
method,
endpoint,
json=json,
params=params,
headers=headers,
**kwargs,
)
# Log response if logging is enabled
if self.enable_logging:
self.logger.info(f"Received response: {response.status_code}")
return response
# Use the retry mechanism from base client
request_context = f"{method} {endpoint}"
response = self._retry_request(make_request, request_context)
# Handle error responses (API errors don't retry)
self._handle_error_response(response)
# httpx.Client automatically prepends base_url
response = self._client.request(
method,
endpoint,
json=json,
params=params,
headers=headers,
**kwargs,
)
return response
def _handle_error_response(self, response, is_upload_request: bool = False) -> None:
"""Handle HTTP error responses and raise appropriate exceptions."""
if response.status_code < 400:
return # Success response
try:
error_data = response.json()
message = error_data.get("message", f"HTTP {response.status_code}")
except (ValueError, KeyError):
message = f"HTTP {response.status_code}"
error_data = None
# Log error response if logging is enabled
if self.enable_logging:
self.logger.error(f"API error: {response.status_code} - {message}")
if response.status_code == 401:
raise AuthenticationError(message, response.status_code, error_data)
elif response.status_code == 429:
retry_after = response.headers.get("Retry-After")
raise RateLimitError(message, retry_after)
elif response.status_code == 422:
raise ValidationError(message, response.status_code, error_data)
elif response.status_code == 400:
# Check if this is a file upload error based on the URL or context
current_url = getattr(response, "url", "") or ""
if is_upload_request or "upload" in str(current_url).lower() or "files" in str(current_url).lower():
raise FileUploadError(message, response.status_code, error_data)
else:
raise APIError(message, response.status_code, error_data)
elif response.status_code >= 500:
# Server errors should raise APIError
raise APIError(message, response.status_code, error_data)
elif response.status_code >= 400:
raise APIError(message, response.status_code, error_data)
def _send_request_with_files(self, method: str, endpoint: str, data: dict, files: dict):
"""Send an HTTP request with file uploads.
@@ -186,12 +102,6 @@ class DifyClient(BaseClientMixin):
"""
headers = {"Authorization": f"Bearer {self.api_key}"}
# Log file upload request if logging is enabled
if self.enable_logging:
self.logger.info(f"Sending {method} file upload request to {endpoint}")
self.logger.debug(f"Form data: {data}")
self.logger.debug(f"Files: {files}")
response = self._client.request(
method,
endpoint,
@@ -200,17 +110,9 @@ class DifyClient(BaseClientMixin):
files=files,
)
# Log response if logging is enabled
if self.enable_logging:
self.logger.info(f"Received file upload response: {response.status_code}")
# Handle error responses
self._handle_error_response(response, is_upload_request=True)
return response
def message_feedback(self, message_id: str, rating: Literal["like", "dislike"], user: str):
self._validate_params(message_id=message_id, rating=rating, user=user)
data = {"rating": rating, "user": user}
return self._send_request("POST", f"/messages/{message_id}/feedbacks", data)
@@ -242,72 +144,6 @@ class DifyClient(BaseClientMixin):
"""Get file preview by file ID."""
return self._send_request("GET", f"/files/{file_id}/preview")
# App Configuration APIs
def get_app_site_config(self, app_id: str):
"""Get app site configuration.
Args:
app_id: ID of the app
Returns:
App site configuration
"""
url = f"/apps/{app_id}/site/config"
return self._send_request("GET", url)
def update_app_site_config(self, app_id: str, config_data: Dict[str, Any]):
"""Update app site configuration.
Args:
app_id: ID of the app
config_data: Configuration data to update
Returns:
Updated app site configuration
"""
url = f"/apps/{app_id}/site/config"
return self._send_request("PUT", url, json=config_data)
def get_app_api_tokens(self, app_id: str):
"""Get API tokens for an app.
Args:
app_id: ID of the app
Returns:
List of API tokens
"""
url = f"/apps/{app_id}/api-tokens"
return self._send_request("GET", url)
def create_app_api_token(self, app_id: str, name: str, description: str | None = None):
"""Create a new API token for an app.
Args:
app_id: ID of the app
name: Name for the API token
description: Description for the API token (optional)
Returns:
Created API token information
"""
data = {"name": name, "description": description}
url = f"/apps/{app_id}/api-tokens"
return self._send_request("POST", url, json=data)
def delete_app_api_token(self, app_id: str, token_id: str):
"""Delete an API token.
Args:
app_id: ID of the app
token_id: ID of the token to delete
Returns:
Deletion result
"""
url = f"/apps/{app_id}/api-tokens/{token_id}"
return self._send_request("DELETE", url)
class CompletionClient(DifyClient):
def create_completion_message(
@@ -315,16 +151,8 @@ class CompletionClient(DifyClient):
inputs: dict,
response_mode: Literal["blocking", "streaming"],
user: str,
files: Dict[str, Any] | None = None,
files: dict | None = None,
):
# Validate parameters
if not isinstance(inputs, dict):
raise ValidationError("inputs must be a dictionary")
if response_mode not in ["blocking", "streaming"]:
raise ValidationError("response_mode must be 'blocking' or 'streaming'")
self._validate_params(inputs=inputs, response_mode=response_mode, user=user)
data = {
"inputs": inputs,
"response_mode": response_mode,
@@ -347,18 +175,8 @@ class ChatClient(DifyClient):
user: str,
response_mode: Literal["blocking", "streaming"] = "blocking",
conversation_id: str | None = None,
files: Dict[str, Any] | None = None,
files: dict | None = None,
):
# Validate parameters
if not isinstance(inputs, dict):
raise ValidationError("inputs must be a dictionary")
if not isinstance(query, str) or not query.strip():
raise ValidationError("query must be a non-empty string")
if response_mode not in ["blocking", "streaming"]:
raise ValidationError("response_mode must be 'blocking' or 'streaming'")
self._validate_params(inputs=inputs, query=query, user=user, response_mode=response_mode)
data = {
"inputs": inputs,
"query": query,
@@ -420,7 +238,7 @@ class ChatClient(DifyClient):
data = {"user": user}
return self._send_request("DELETE", f"/conversations/{conversation_id}", data)
def audio_to_text(self, audio_file: Union[IO[bytes], tuple], user: str):
def audio_to_text(self, audio_file: IO[bytes] | tuple, user: str):
data = {"user": user}
files = {"file": audio_file}
return self._send_request_with_files("POST", "/audio-to-text", data, files)
@@ -495,48 +313,7 @@ class ChatClient(DifyClient):
"""
data = {"value": value, "user": user}
url = f"/conversations/{conversation_id}/variables/{variable_id}"
return self._send_request("PUT", url, json=data)
def delete_annotation_with_response(self, annotation_id: str):
"""Delete an annotation with full response handling."""
url = f"/apps/annotations/{annotation_id}"
return self._send_request("DELETE", url)
def list_conversation_variables_with_pagination(
self, conversation_id: str, user: str, page: int = 1, limit: int = 20
):
"""List conversation variables with pagination."""
params = {"page": page, "limit": limit, "user": user}
url = f"/conversations/{conversation_id}/variables"
return self._send_request("GET", url, params=params)
def update_conversation_variable_with_response(self, conversation_id: str, variable_id: str, user: str, value: Any):
"""Update a conversation variable with full response handling."""
data = {"value": value, "user": user}
url = f"/conversations/{conversation_id}/variables/{variable_id}"
return self._send_request("PUT", url, json=data)
# Enhanced Annotation APIs
def get_annotation_reply_job_status(self, action: str, job_id: str):
"""Get status of an annotation reply action job."""
url = f"/apps/annotation-reply/{action}/status/{job_id}"
return self._send_request("GET", url)
def list_annotations_with_pagination(self, page: int = 1, limit: int = 20, keyword: str | None = None):
"""List annotations with pagination."""
params = {"page": page, "limit": limit, "keyword": keyword}
return self._send_request("GET", "/apps/annotations", params=params)
def create_annotation_with_response(self, question: str, answer: str):
"""Create an annotation with full response handling."""
data = {"question": question, "answer": answer}
return self._send_request("POST", "/apps/annotations", json=data)
def update_annotation_with_response(self, annotation_id: str, question: str, answer: str):
"""Update an annotation with full response handling."""
data = {"question": question, "answer": answer}
url = f"/apps/annotations/{annotation_id}"
return self._send_request("PUT", url, json=data)
return self._send_request("PATCH", url, json=data)
class WorkflowClient(DifyClient):
@@ -599,68 +376,6 @@ class WorkflowClient(DifyClient):
stream=(response_mode == "streaming"),
)
# Enhanced Workflow APIs
def get_workflow_draft(self, app_id: str):
"""Get workflow draft configuration.
Args:
app_id: ID of the workflow app
Returns:
Workflow draft configuration
"""
url = f"/apps/{app_id}/workflow/draft"
return self._send_request("GET", url)
def update_workflow_draft(self, app_id: str, workflow_data: Dict[str, Any]):
"""Update workflow draft configuration.
Args:
app_id: ID of the workflow app
workflow_data: Workflow configuration data
Returns:
Updated workflow draft
"""
url = f"/apps/{app_id}/workflow/draft"
return self._send_request("PUT", url, json=workflow_data)
def publish_workflow(self, app_id: str):
"""Publish workflow from draft.
Args:
app_id: ID of the workflow app
Returns:
Published workflow information
"""
url = f"/apps/{app_id}/workflow/publish"
return self._send_request("POST", url)
def get_workflow_run_history(
self,
app_id: str,
page: int = 1,
limit: int = 20,
status: Literal["succeeded", "failed", "stopped"] | None = None,
):
"""Get workflow run history.
Args:
app_id: ID of the workflow app
page: Page number (default: 1)
limit: Number of items per page (default: 20)
status: Filter by status (optional)
Returns:
Paginated workflow run history
"""
params = {"page": page, "limit": limit}
if status:
params["status"] = status
url = f"/apps/{app_id}/workflow/runs"
return self._send_request("GET", url, params=params)
class WorkspaceClient(DifyClient):
"""Client for workspace-related operations."""
@@ -670,41 +385,6 @@ class WorkspaceClient(DifyClient):
url = f"/workspaces/current/models/model-types/{model_type}"
return self._send_request("GET", url)
def get_available_models_by_type(self, model_type: str):
"""Get available models by model type (enhanced version)."""
url = f"/workspaces/current/models/model-types/{model_type}"
return self._send_request("GET", url)
def get_model_providers(self):
"""Get all model providers."""
return self._send_request("GET", "/workspaces/current/model-providers")
def get_model_provider_models(self, provider_name: str):
"""Get models for a specific provider."""
url = f"/workspaces/current/model-providers/{provider_name}/models"
return self._send_request("GET", url)
def validate_model_provider_credentials(self, provider_name: str, credentials: Dict[str, Any]):
"""Validate model provider credentials."""
url = f"/workspaces/current/model-providers/{provider_name}/credentials/validate"
return self._send_request("POST", url, json=credentials)
# File Management APIs
def get_file_info(self, file_id: str):
"""Get information about a specific file."""
url = f"/files/{file_id}/info"
return self._send_request("GET", url)
def get_file_download_url(self, file_id: str):
"""Get download URL for a file."""
url = f"/files/{file_id}/download-url"
return self._send_request("GET", url)
def delete_file(self, file_id: str):
"""Delete a file."""
url = f"/files/{file_id}"
return self._send_request("DELETE", url)
class KnowledgeBaseClient(DifyClient):
def __init__(
@@ -736,7 +416,7 @@ class KnowledgeBaseClient(DifyClient):
def list_datasets(self, page: int = 1, page_size: int = 20, **kwargs):
return self._send_request("GET", "/datasets", params={"page": page, "limit": page_size}, **kwargs)
def create_document_by_text(self, name, text, extra_params: Dict[str, Any] | None = None, **kwargs):
def create_document_by_text(self, name, text, extra_params: dict | None = None, **kwargs):
"""
Create a document by text.
@@ -778,7 +458,7 @@ class KnowledgeBaseClient(DifyClient):
document_id: str,
name: str,
text: str,
extra_params: Dict[str, Any] | None = None,
extra_params: dict | None = None,
**kwargs,
):
"""
@@ -817,7 +497,7 @@ class KnowledgeBaseClient(DifyClient):
self,
file_path: str,
original_document_id: str | None = None,
extra_params: Dict[str, Any] | None = None,
extra_params: dict | None = None,
):
"""
Create a document by file.
@@ -857,12 +537,7 @@ class KnowledgeBaseClient(DifyClient):
url = f"/datasets/{self._get_dataset_id()}/document/create_by_file"
return self._send_request_with_files("POST", url, {"data": json.dumps(data)}, files)
def update_document_by_file(
self,
document_id: str,
file_path: str,
extra_params: Dict[str, Any] | None = None,
):
def update_document_by_file(self, document_id: str, file_path: str, extra_params: dict | None = None):
"""
Update a document by file.
@@ -1218,50 +893,3 @@ class KnowledgeBaseClient(DifyClient):
url = f"/datasets/{ds_id}/documents/status/{action}"
data = {"document_ids": document_ids}
return self._send_request("PATCH", url, json=data)
# Enhanced Dataset APIs
def create_dataset_from_template(self, template_name: str, name: str, description: str | None = None):
"""Create a dataset from a predefined template.
Args:
template_name: Name of the template to use
name: Name for the new dataset
description: Description for the dataset (optional)
Returns:
Created dataset information
"""
data = {
"template_name": template_name,
"name": name,
"description": description,
}
return self._send_request("POST", "/datasets/from-template", json=data)
def duplicate_dataset(self, dataset_id: str, name: str):
"""Duplicate an existing dataset.
Args:
dataset_id: ID of dataset to duplicate
name: Name for duplicated dataset
Returns:
New dataset information
"""
data = {"name": name}
url = f"/datasets/{dataset_id}/duplicate"
return self._send_request("POST", url, json=data)
def list_conversation_variables_with_pagination(
self, conversation_id: str, user: str, page: int = 1, limit: int = 20
):
"""List conversation variables with pagination."""
params = {"page": page, "limit": limit, "user": user}
url = f"/conversations/{conversation_id}/variables"
return self._send_request("GET", url, params=params)
def update_conversation_variable_with_response(self, conversation_id: str, variable_id: str, user: str, value: Any):
"""Update a conversation variable with full response handling."""
data = {"value": value, "user": user}
url = f"/conversations/{conversation_id}/variables/{variable_id}"
return self._send_request("PUT", url, json=data)
@@ -1,71 +0,0 @@
"""Custom exceptions for the Dify client."""
from typing import Optional, Dict, Any
class DifyClientError(Exception):
"""Base exception for all Dify client errors."""
def __init__(self, message: str, status_code: int | None = None, response: Dict[str, Any] | None = None):
super().__init__(message)
self.message = message
self.status_code = status_code
self.response = response
class APIError(DifyClientError):
"""Raised when the API returns an error response."""
def __init__(self, message: str, status_code: int, response: Dict[str, Any] | None = None):
super().__init__(message, status_code, response)
self.status_code = status_code
class AuthenticationError(DifyClientError):
"""Raised when authentication fails."""
pass
class RateLimitError(DifyClientError):
"""Raised when rate limit is exceeded."""
def __init__(self, message: str = "Rate limit exceeded", retry_after: int | None = None):
super().__init__(message)
self.retry_after = retry_after
class ValidationError(DifyClientError):
"""Raised when request validation fails."""
pass
class NetworkError(DifyClientError):
"""Raised when network-related errors occur."""
pass
class TimeoutError(DifyClientError):
"""Raised when request times out."""
pass
class FileUploadError(DifyClientError):
"""Raised when file upload fails."""
pass
class DatasetError(DifyClientError):
"""Raised when dataset operations fail."""
pass
class WorkflowError(DifyClientError):
"""Raised when workflow operations fail."""
pass
-396
View File
@@ -1,396 +0,0 @@
"""Response models for the Dify client with proper type hints."""
from typing import Optional, List, Dict, Any, Literal, Union
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class BaseResponse:
"""Base response model."""
success: bool = True
message: str | None = None
@dataclass
class ErrorResponse(BaseResponse):
"""Error response model."""
error_code: str | None = None
details: Dict[str, Any] | None = None
success: bool = False
@dataclass
class FileInfo:
"""File information model."""
id: str
name: str
size: int
mime_type: str
url: str | None = None
created_at: datetime | None = None
@dataclass
class MessageResponse(BaseResponse):
"""Message response model."""
id: str = ""
answer: str = ""
conversation_id: str | None = None
created_at: int | None = None
metadata: Dict[str, Any] | None = None
files: List[Dict[str, Any]] | None = None
@dataclass
class ConversationResponse(BaseResponse):
"""Conversation response model."""
id: str = ""
name: str = ""
inputs: Dict[str, Any] | None = None
status: str | None = None
created_at: int | None = None
updated_at: int | None = None
@dataclass
class DatasetResponse(BaseResponse):
"""Dataset response model."""
id: str = ""
name: str = ""
description: str | None = None
permission: str | None = None
indexing_technique: str | None = None
embedding_model: str | None = None
embedding_model_provider: str | None = None
retrieval_model: Dict[str, Any] | None = None
document_count: int | None = None
word_count: int | None = None
app_count: int | None = None
created_at: int | None = None
updated_at: int | None = None
@dataclass
class DocumentResponse(BaseResponse):
"""Document response model."""
id: str = ""
name: str = ""
data_source_type: str | None = None
data_source_info: Dict[str, Any] | None = None
dataset_process_rule_id: str | None = None
batch: str | None = None
position: int | None = None
enabled: bool | None = None
disabled_at: float | None = None
disabled_by: str | None = None
archived: bool | None = None
archived_reason: str | None = None
archived_at: float | None = None
archived_by: str | None = None
word_count: int | None = None
hit_count: int | None = None
doc_form: str | None = None
doc_metadata: Dict[str, Any] | None = None
created_at: float | None = None
updated_at: float | None = None
indexing_status: str | None = None
completed_at: float | None = None
paused_at: float | None = None
error: str | None = None
stopped_at: float | None = None
@dataclass
class DocumentSegmentResponse(BaseResponse):
"""Document segment response model."""
id: str = ""
position: int | None = None
document_id: str | None = None
content: str | None = None
answer: str | None = None
word_count: int | None = None
tokens: int | None = None
keywords: List[str] | None = None
index_node_id: str | None = None
index_node_hash: str | None = None
hit_count: int | None = None
enabled: bool | None = None
disabled_at: float | None = None
disabled_by: str | None = None
status: str | None = None
created_by: str | None = None
created_at: float | None = None
indexing_at: float | None = None
completed_at: float | None = None
error: str | None = None
stopped_at: float | None = None
@dataclass
class WorkflowRunResponse(BaseResponse):
"""Workflow run response model."""
id: str = ""
workflow_id: str | None = None
status: Literal["running", "succeeded", "failed", "stopped"] | None = None
inputs: Dict[str, Any] | None = None
outputs: Dict[str, Any] | None = None
error: str | None = None
elapsed_time: float | None = None
total_tokens: int | None = None
total_steps: int | None = None
created_at: float | None = None
finished_at: float | None = None
@dataclass
class ApplicationParametersResponse(BaseResponse):
"""Application parameters response model."""
opening_statement: str | None = None
suggested_questions: List[str] | None = None
speech_to_text: Dict[str, Any] | None = None
text_to_speech: Dict[str, Any] | None = None
retriever_resource: Dict[str, Any] | None = None
sensitive_word_avoidance: Dict[str, Any] | None = None
file_upload: Dict[str, Any] | None = None
system_parameters: Dict[str, Any] | None = None
user_input_form: List[Dict[str, Any]] | None = None
@dataclass
class AnnotationResponse(BaseResponse):
"""Annotation response model."""
id: str = ""
question: str = ""
answer: str = ""
content: str | None = None
created_at: float | None = None
updated_at: float | None = None
created_by: str | None = None
updated_by: str | None = None
hit_count: int | None = None
@dataclass
class PaginatedResponse(BaseResponse):
"""Paginated response model."""
data: List[Any] = field(default_factory=list)
has_more: bool = False
limit: int = 0
total: int = 0
page: int | None = None
@dataclass
class ConversationVariableResponse(BaseResponse):
"""Conversation variable response model."""
conversation_id: str = ""
variables: List[Dict[str, Any]] = field(default_factory=list)
@dataclass
class FileUploadResponse(BaseResponse):
"""File upload response model."""
id: str = ""
name: str = ""
size: int = 0
mime_type: str = ""
url: str | None = None
created_at: float | None = None
@dataclass
class AudioResponse(BaseResponse):
"""Audio generation/response model."""
audio: str | None = None # Base64 encoded audio data or URL
audio_url: str | None = None
duration: float | None = None
sample_rate: int | None = None
@dataclass
class SuggestedQuestionsResponse(BaseResponse):
"""Suggested questions response model."""
message_id: str = ""
questions: List[str] = field(default_factory=list)
@dataclass
class AppInfoResponse(BaseResponse):
"""App info response model."""
id: str = ""
name: str = ""
description: str | None = None
icon: str | None = None
icon_background: str | None = None
mode: str | None = None
tags: List[str] | None = None
enable_site: bool | None = None
enable_api: bool | None = None
api_token: str | None = None
@dataclass
class WorkspaceModelsResponse(BaseResponse):
"""Workspace models response model."""
models: List[Dict[str, Any]] = field(default_factory=list)
@dataclass
class HitTestingResponse(BaseResponse):
"""Hit testing response model."""
query: str = ""
records: List[Dict[str, Any]] = field(default_factory=list)
@dataclass
class DatasetTagsResponse(BaseResponse):
"""Dataset tags response model."""
tags: List[Dict[str, Any]] = field(default_factory=list)
@dataclass
class WorkflowLogsResponse(BaseResponse):
"""Workflow logs response model."""
logs: List[Dict[str, Any]] = field(default_factory=list)
total: int = 0
page: int = 0
limit: int = 0
has_more: bool = False
@dataclass
class ModelProviderResponse(BaseResponse):
"""Model provider response model."""
provider_name: str = ""
provider_type: str = ""
models: List[Dict[str, Any]] = field(default_factory=list)
is_enabled: bool = False
credentials: Dict[str, Any] | None = None
@dataclass
class FileInfoResponse(BaseResponse):
"""File info response model."""
id: str = ""
name: str = ""
size: int = 0
mime_type: str = ""
url: str | None = None
created_at: int | None = None
metadata: Dict[str, Any] | None = None
@dataclass
class WorkflowDraftResponse(BaseResponse):
"""Workflow draft response model."""
id: str = ""
app_id: str = ""
draft_data: Dict[str, Any] = field(default_factory=dict)
version: int = 0
created_at: int | None = None
updated_at: int | None = None
@dataclass
class ApiTokenResponse(BaseResponse):
"""API token response model."""
id: str = ""
name: str = ""
token: str = ""
description: str | None = None
created_at: int | None = None
last_used_at: int | None = None
is_active: bool = True
@dataclass
class JobStatusResponse(BaseResponse):
"""Job status response model."""
job_id: str = ""
job_status: str = ""
error_msg: str | None = None
progress: float | None = None
created_at: int | None = None
updated_at: int | None = None
@dataclass
class DatasetQueryResponse(BaseResponse):
"""Dataset query response model."""
query: str = ""
records: List[Dict[str, Any]] = field(default_factory=list)
total: int = 0
search_time: float | None = None
retrieval_model: Dict[str, Any] | None = None
@dataclass
class DatasetTemplateResponse(BaseResponse):
"""Dataset template response model."""
template_name: str = ""
display_name: str = ""
description: str = ""
category: str = ""
icon: str | None = None
config_schema: Dict[str, Any] = field(default_factory=dict)
# Type aliases for common response types
ResponseType = Union[
BaseResponse,
ErrorResponse,
MessageResponse,
ConversationResponse,
DatasetResponse,
DocumentResponse,
DocumentSegmentResponse,
WorkflowRunResponse,
ApplicationParametersResponse,
AnnotationResponse,
PaginatedResponse,
ConversationVariableResponse,
FileUploadResponse,
AudioResponse,
SuggestedQuestionsResponse,
AppInfoResponse,
WorkspaceModelsResponse,
HitTestingResponse,
DatasetTagsResponse,
WorkflowLogsResponse,
ModelProviderResponse,
FileInfoResponse,
WorkflowDraftResponse,
ApiTokenResponse,
JobStatusResponse,
DatasetQueryResponse,
DatasetTemplateResponse,
]
@@ -1,264 +0,0 @@
"""
Advanced usage examples for the Dify Python SDK.
This example demonstrates:
- Error handling and retries
- Logging configuration
- Context managers
- Async usage
- File uploads
- Dataset management
"""
import asyncio
import logging
from pathlib import Path
from dify_client import (
ChatClient,
CompletionClient,
AsyncChatClient,
KnowledgeBaseClient,
DifyClient,
)
from dify_client.exceptions import (
APIError,
RateLimitError,
AuthenticationError,
DifyClientError,
)
def setup_logging():
"""Setup logging for the SDK."""
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
def example_chat_with_error_handling():
"""Example of chat with comprehensive error handling."""
api_key = "your-api-key-here"
try:
with ChatClient(api_key, enable_logging=True) as client:
# Simple chat message
response = client.create_chat_message(
inputs={}, query="Hello, how are you?", user="user-123", response_mode="blocking"
)
result = response.json()
print(f"Response: {result.get('answer')}")
except AuthenticationError as e:
print(f"Authentication failed: {e}")
print("Please check your API key")
except RateLimitError as e:
print(f"Rate limit exceeded: {e}")
if e.retry_after:
print(f"Retry after {e.retry_after} seconds")
except APIError as e:
print(f"API error: {e.message}")
print(f"Status code: {e.status_code}")
except DifyClientError as e:
print(f"Dify client error: {e}")
except Exception as e:
print(f"Unexpected error: {e}")
def example_completion_with_files():
"""Example of completion with file upload."""
api_key = "your-api-key-here"
with CompletionClient(api_key) as client:
# Upload an image file first
file_path = "path/to/your/image.jpg"
try:
with open(file_path, "rb") as f:
files = {"file": (Path(file_path).name, f, "image/jpeg")}
upload_response = client.file_upload("user-123", files)
upload_response.raise_for_status()
file_id = upload_response.json().get("id")
print(f"File uploaded with ID: {file_id}")
# Use the uploaded file in completion
files_list = [{"type": "image", "transfer_method": "local_file", "upload_file_id": file_id}]
completion_response = client.create_completion_message(
inputs={"query": "Describe this image"}, response_mode="blocking", user="user-123", files=files_list
)
result = completion_response.json()
print(f"Completion result: {result.get('answer')}")
except FileNotFoundError:
print(f"File not found: {file_path}")
except Exception as e:
print(f"Error during file upload/completion: {e}")
def example_dataset_management():
"""Example of dataset management operations."""
api_key = "your-api-key-here"
with KnowledgeBaseClient(api_key) as kb_client:
try:
# Create a new dataset
create_response = kb_client.create_dataset(name="My Test Dataset")
create_response.raise_for_status()
dataset_id = create_response.json().get("id")
print(f"Created dataset with ID: {dataset_id}")
# Create a client with the dataset ID
dataset_client = KnowledgeBaseClient(api_key, dataset_id=dataset_id)
# Add a document by text
doc_response = dataset_client.create_document_by_text(
name="Test Document", text="This is a test document for the knowledge base."
)
doc_response.raise_for_status()
document_id = doc_response.json().get("document", {}).get("id")
print(f"Created document with ID: {document_id}")
# List documents
list_response = dataset_client.list_documents()
list_response.raise_for_status()
documents = list_response.json().get("data", [])
print(f"Dataset contains {len(documents)} documents")
# Update dataset configuration
update_response = dataset_client.update_dataset(
name="Updated Dataset Name", description="Updated description", indexing_technique="high_quality"
)
update_response.raise_for_status()
print("Dataset updated successfully")
except Exception as e:
print(f"Dataset management error: {e}")
async def example_async_chat():
"""Example of async chat usage."""
api_key = "your-api-key-here"
try:
async with AsyncChatClient(api_key) as client:
# Create chat message
response = await client.create_chat_message(
inputs={}, query="What's the weather like?", user="user-456", response_mode="blocking"
)
result = response.json()
print(f"Async response: {result.get('answer')}")
# Get conversations
conversations = await client.get_conversations("user-456")
conversations.raise_for_status()
conv_data = conversations.json()
print(f"Found {len(conv_data.get('data', []))} conversations")
except Exception as e:
print(f"Async chat error: {e}")
def example_streaming_response():
"""Example of handling streaming responses."""
api_key = "your-api-key-here"
with ChatClient(api_key) as client:
try:
response = client.create_chat_message(
inputs={}, query="Tell me a story", user="user-789", response_mode="streaming"
)
print("Streaming response:")
for line in response.iter_lines(decode_unicode=True):
if line.startswith("data:"):
data = line[5:].strip()
if data:
import json
try:
chunk = json.loads(data)
answer = chunk.get("answer", "")
if answer:
print(answer, end="", flush=True)
except json.JSONDecodeError:
continue
print() # New line after streaming
except Exception as e:
print(f"Streaming error: {e}")
def example_application_info():
"""Example of getting application information."""
api_key = "your-api-key-here"
with DifyClient(api_key) as client:
try:
# Get app info
info_response = client.get_app_info()
info_response.raise_for_status()
app_info = info_response.json()
print(f"App name: {app_info.get('name')}")
print(f"App mode: {app_info.get('mode')}")
print(f"App tags: {app_info.get('tags', [])}")
# Get app parameters
params_response = client.get_application_parameters("user-123")
params_response.raise_for_status()
params = params_response.json()
print(f"Opening statement: {params.get('opening_statement')}")
print(f"Suggested questions: {params.get('suggested_questions', [])}")
except Exception as e:
print(f"App info error: {e}")
def main():
"""Run all examples."""
setup_logging()
print("=== Dify Python SDK Advanced Usage Examples ===\n")
print("1. Chat with Error Handling:")
example_chat_with_error_handling()
print()
print("2. Completion with Files:")
example_completion_with_files()
print()
print("3. Dataset Management:")
example_dataset_management()
print()
print("4. Async Chat:")
asyncio.run(example_async_chat())
print()
print("5. Streaming Response:")
example_streaming_response()
print()
print("6. Application Info:")
example_application_info()
print()
print("All examples completed!")
if __name__ == "__main__":
main()
+1 -1
View File
@@ -5,7 +5,7 @@ description = "A package for interacting with the Dify Service-API"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
"httpx[http2]>=0.27.0",
"httpx>=0.27.0",
"aiofiles>=23.0.0",
]
authors = [
+69 -308
View File
@@ -1,7 +1,6 @@
import os
import time
import unittest
from unittest.mock import Mock, patch, mock_open
from dify_client.client import (
ChatClient,
@@ -18,46 +17,38 @@ FILE_PATH_BASE = os.path.dirname(__file__)
class TestKnowledgeBaseClient(unittest.TestCase):
def setUp(self):
self.api_key = "test-api-key"
self.base_url = "https://api.dify.ai/v1"
self.knowledge_base_client = KnowledgeBaseClient(self.api_key, base_url=self.base_url)
self.knowledge_base_client = KnowledgeBaseClient(API_KEY, base_url=API_BASE_URL)
self.README_FILE_PATH = os.path.abspath(os.path.join(FILE_PATH_BASE, "../README.md"))
self.dataset_id = "test-dataset-id"
self.document_id = "test-document-id"
self.segment_id = "test-segment-id"
self.batch_id = "test-batch-id"
self.dataset_id = None
self.document_id = None
self.segment_id = None
self.batch_id = None
def _get_dataset_kb_client(self):
return KnowledgeBaseClient(self.api_key, base_url=self.base_url, dataset_id=self.dataset_id)
@patch("dify_client.client.httpx.Client")
def test_001_create_dataset(self, mock_httpx_client):
# Mock the HTTP response
mock_response = Mock()
mock_response.json.return_value = {"id": self.dataset_id, "name": "test_dataset"}
mock_response.status_code = 200
mock_client_instance = Mock()
mock_client_instance.request.return_value = mock_response
mock_httpx_client.return_value = mock_client_instance
# Re-create client with mocked httpx
self.knowledge_base_client = KnowledgeBaseClient(self.api_key, base_url=self.base_url)
self.assertIsNotNone(self.dataset_id)
return KnowledgeBaseClient(API_KEY, base_url=API_BASE_URL, dataset_id=self.dataset_id)
def test_001_create_dataset(self):
response = self.knowledge_base_client.create_dataset(name="test_dataset")
data = response.json()
self.assertIn("id", data)
self.dataset_id = data["id"]
self.assertEqual("test_dataset", data["name"])
# the following tests require to be executed in order because they use
# the dataset/document/segment ids from the previous test
self._test_002_list_datasets()
self._test_003_create_document_by_text()
time.sleep(1)
self._test_004_update_document_by_text()
# self._test_005_batch_indexing_status()
time.sleep(1)
self._test_006_update_document_by_file()
time.sleep(1)
self._test_007_list_documents()
self._test_008_delete_document()
self._test_009_create_document_by_file()
time.sleep(1)
self._test_010_add_segments()
self._test_011_query_segments()
self._test_012_update_document_segment()
@@ -65,12 +56,6 @@ class TestKnowledgeBaseClient(unittest.TestCase):
self._test_014_delete_dataset()
def _test_002_list_datasets(self):
# Mock the response - using the already mocked client from test_001_create_dataset
mock_response = Mock()
mock_response.json.return_value = {"data": [], "total": 0}
mock_response.status_code = 200
self.knowledge_base_client._client.request.return_value = mock_response
response = self.knowledge_base_client.list_datasets()
data = response.json()
self.assertIn("data", data)
@@ -78,62 +63,45 @@ class TestKnowledgeBaseClient(unittest.TestCase):
def _test_003_create_document_by_text(self):
client = self._get_dataset_kb_client()
# Mock the response
mock_response = Mock()
mock_response.json.return_value = {"document": {"id": self.document_id}, "batch": self.batch_id}
mock_response.status_code = 200
client._client.request.return_value = mock_response
response = client.create_document_by_text("test_document", "test_text")
data = response.json()
self.assertIn("document", data)
self.document_id = data["document"]["id"]
self.batch_id = data["batch"]
def _test_004_update_document_by_text(self):
client = self._get_dataset_kb_client()
# Mock the response
mock_response = Mock()
mock_response.json.return_value = {"document": {"id": self.document_id}, "batch": self.batch_id}
mock_response.status_code = 200
client._client.request.return_value = mock_response
self.assertIsNotNone(self.document_id)
response = client.update_document_by_text(self.document_id, "test_document_updated", "test_text_updated")
data = response.json()
self.assertIn("document", data)
self.assertIn("batch", data)
self.batch_id = data["batch"]
def _test_005_batch_indexing_status(self):
client = self._get_dataset_kb_client()
response = client.batch_indexing_status(self.batch_id)
response.json()
self.assertEqual(response.status_code, 200)
def _test_006_update_document_by_file(self):
client = self._get_dataset_kb_client()
# Mock the response
mock_response = Mock()
mock_response.json.return_value = {"document": {"id": self.document_id}, "batch": self.batch_id}
mock_response.status_code = 200
client._client.request.return_value = mock_response
self.assertIsNotNone(self.document_id)
response = client.update_document_by_file(self.document_id, self.README_FILE_PATH)
data = response.json()
self.assertIn("document", data)
self.assertIn("batch", data)
self.batch_id = data["batch"]
def _test_007_list_documents(self):
client = self._get_dataset_kb_client()
# Mock the response
mock_response = Mock()
mock_response.json.return_value = {"data": []}
mock_response.status_code = 200
client._client.request.return_value = mock_response
response = client.list_documents()
data = response.json()
self.assertIn("data", data)
def _test_008_delete_document(self):
client = self._get_dataset_kb_client()
# Mock the response
mock_response = Mock()
mock_response.json.return_value = {"result": "success"}
mock_response.status_code = 200
client._client.request.return_value = mock_response
self.assertIsNotNone(self.document_id)
response = client.delete_document(self.document_id)
data = response.json()
self.assertIn("result", data)
@@ -141,37 +109,23 @@ class TestKnowledgeBaseClient(unittest.TestCase):
def _test_009_create_document_by_file(self):
client = self._get_dataset_kb_client()
# Mock the response
mock_response = Mock()
mock_response.json.return_value = {"document": {"id": self.document_id}, "batch": self.batch_id}
mock_response.status_code = 200
client._client.request.return_value = mock_response
response = client.create_document_by_file(self.README_FILE_PATH)
data = response.json()
self.assertIn("document", data)
self.document_id = data["document"]["id"]
self.batch_id = data["batch"]
def _test_010_add_segments(self):
client = self._get_dataset_kb_client()
# Mock the response
mock_response = Mock()
mock_response.json.return_value = {"data": [{"id": self.segment_id, "content": "test text segment 1"}]}
mock_response.status_code = 200
client._client.request.return_value = mock_response
response = client.add_segments(self.document_id, [{"content": "test text segment 1"}])
data = response.json()
self.assertIn("data", data)
self.assertGreater(len(data["data"]), 0)
segment = data["data"][0]
self.segment_id = segment["id"]
def _test_011_query_segments(self):
client = self._get_dataset_kb_client()
# Mock the response
mock_response = Mock()
mock_response.json.return_value = {"data": [{"id": self.segment_id, "content": "test text segment 1"}]}
mock_response.status_code = 200
client._client.request.return_value = mock_response
response = client.query_segments(self.document_id)
data = response.json()
self.assertIn("data", data)
@@ -179,12 +133,7 @@ class TestKnowledgeBaseClient(unittest.TestCase):
def _test_012_update_document_segment(self):
client = self._get_dataset_kb_client()
# Mock the response
mock_response = Mock()
mock_response.json.return_value = {"data": {"id": self.segment_id, "content": "test text segment 1 updated"}}
mock_response.status_code = 200
client._client.request.return_value = mock_response
self.assertIsNotNone(self.segment_id)
response = client.update_document_segment(
self.document_id,
self.segment_id,
@@ -192,16 +141,13 @@ class TestKnowledgeBaseClient(unittest.TestCase):
)
data = response.json()
self.assertIn("data", data)
self.assertEqual("test text segment 1 updated", data["data"]["content"])
self.assertGreater(len(data["data"]), 0)
segment = data["data"]
self.assertEqual("test text segment 1 updated", segment["content"])
def _test_013_delete_document_segment(self):
client = self._get_dataset_kb_client()
# Mock the response
mock_response = Mock()
mock_response.json.return_value = {"result": "success"}
mock_response.status_code = 200
client._client.request.return_value = mock_response
self.assertIsNotNone(self.segment_id)
response = client.delete_document_segment(self.document_id, self.segment_id)
data = response.json()
self.assertIn("result", data)
@@ -209,279 +155,94 @@ class TestKnowledgeBaseClient(unittest.TestCase):
def _test_014_delete_dataset(self):
client = self._get_dataset_kb_client()
# Mock the response
mock_response = Mock()
mock_response.status_code = 204
client._client.request.return_value = mock_response
response = client.delete_dataset()
self.assertEqual(204, response.status_code)
class TestChatClient(unittest.TestCase):
@patch("dify_client.client.httpx.Client")
def setUp(self, mock_httpx_client):
self.api_key = "test-api-key"
self.chat_client = ChatClient(self.api_key)
def setUp(self):
self.chat_client = ChatClient(API_KEY)
# Set up default mock response for the client
mock_response = Mock()
mock_response.text = '{"answer": "Hello! This is a test response."}'
mock_response.json.return_value = {"answer": "Hello! This is a test response."}
mock_response.status_code = 200
mock_client_instance = Mock()
mock_client_instance.request.return_value = mock_response
mock_httpx_client.return_value = mock_client_instance
@patch("dify_client.client.httpx.Client")
def test_create_chat_message(self, mock_httpx_client):
# Mock the HTTP response
mock_response = Mock()
mock_response.text = '{"answer": "Hello! This is a test response."}'
mock_response.json.return_value = {"answer": "Hello! This is a test response."}
mock_response.status_code = 200
mock_client_instance = Mock()
mock_client_instance.request.return_value = mock_response
mock_httpx_client.return_value = mock_client_instance
# Create client with mocked httpx
chat_client = ChatClient(self.api_key)
response = chat_client.create_chat_message({}, "Hello, World!", "test_user")
def test_create_chat_message(self):
response = self.chat_client.create_chat_message({}, "Hello, World!", "test_user")
self.assertIn("answer", response.text)
@patch("dify_client.client.httpx.Client")
def test_create_chat_message_with_vision_model_by_remote_url(self, mock_httpx_client):
# Mock the HTTP response
mock_response = Mock()
mock_response.text = '{"answer": "I can see this is a test image description."}'
mock_response.json.return_value = {"answer": "I can see this is a test image description."}
mock_response.status_code = 200
mock_client_instance = Mock()
mock_client_instance.request.return_value = mock_response
mock_httpx_client.return_value = mock_client_instance
# Create client with mocked httpx
chat_client = ChatClient(self.api_key)
files = [{"type": "image", "transfer_method": "remote_url", "url": "https://example.com/test-image.jpg"}]
response = chat_client.create_chat_message({}, "Describe the picture.", "test_user", files=files)
def test_create_chat_message_with_vision_model_by_remote_url(self):
files = [{"type": "image", "transfer_method": "remote_url", "url": "your_image_url"}]
response = self.chat_client.create_chat_message({}, "Describe the picture.", "test_user", files=files)
self.assertIn("answer", response.text)
@patch("dify_client.client.httpx.Client")
def test_create_chat_message_with_vision_model_by_local_file(self, mock_httpx_client):
# Mock the HTTP response
mock_response = Mock()
mock_response.text = '{"answer": "I can see this is a test uploaded image."}'
mock_response.json.return_value = {"answer": "I can see this is a test uploaded image."}
mock_response.status_code = 200
mock_client_instance = Mock()
mock_client_instance.request.return_value = mock_response
mock_httpx_client.return_value = mock_client_instance
# Create client with mocked httpx
chat_client = ChatClient(self.api_key)
def test_create_chat_message_with_vision_model_by_local_file(self):
files = [
{
"type": "image",
"transfer_method": "local_file",
"upload_file_id": "test-file-id",
"upload_file_id": "your_file_id",
}
]
response = chat_client.create_chat_message({}, "Describe the picture.", "test_user", files=files)
response = self.chat_client.create_chat_message({}, "Describe the picture.", "test_user", files=files)
self.assertIn("answer", response.text)
@patch("dify_client.client.httpx.Client")
def test_get_conversation_messages(self, mock_httpx_client):
# Mock the HTTP response
mock_response = Mock()
mock_response.text = '{"answer": "Here are the conversation messages."}'
mock_response.json.return_value = {"answer": "Here are the conversation messages."}
mock_response.status_code = 200
mock_client_instance = Mock()
mock_client_instance.request.return_value = mock_response
mock_httpx_client.return_value = mock_client_instance
# Create client with mocked httpx
chat_client = ChatClient(self.api_key)
response = chat_client.get_conversation_messages("test_user", "test-conversation-id")
def test_get_conversation_messages(self):
response = self.chat_client.get_conversation_messages("test_user", "your_conversation_id")
self.assertIn("answer", response.text)
@patch("dify_client.client.httpx.Client")
def test_get_conversations(self, mock_httpx_client):
# Mock the HTTP response
mock_response = Mock()
mock_response.text = '{"data": [{"id": "conv1", "name": "Test Conversation"}]}'
mock_response.json.return_value = {"data": [{"id": "conv1", "name": "Test Conversation"}]}
mock_response.status_code = 200
mock_client_instance = Mock()
mock_client_instance.request.return_value = mock_response
mock_httpx_client.return_value = mock_client_instance
# Create client with mocked httpx
chat_client = ChatClient(self.api_key)
response = chat_client.get_conversations("test_user")
def test_get_conversations(self):
response = self.chat_client.get_conversations("test_user")
self.assertIn("data", response.text)
class TestCompletionClient(unittest.TestCase):
@patch("dify_client.client.httpx.Client")
def setUp(self, mock_httpx_client):
self.api_key = "test-api-key"
self.completion_client = CompletionClient(self.api_key)
def setUp(self):
self.completion_client = CompletionClient(API_KEY)
# Set up default mock response for the client
mock_response = Mock()
mock_response.text = '{"answer": "This is a test completion response."}'
mock_response.json.return_value = {"answer": "This is a test completion response."}
mock_response.status_code = 200
mock_client_instance = Mock()
mock_client_instance.request.return_value = mock_response
mock_httpx_client.return_value = mock_client_instance
@patch("dify_client.client.httpx.Client")
def test_create_completion_message(self, mock_httpx_client):
# Mock the HTTP response
mock_response = Mock()
mock_response.text = '{"answer": "The weather today is sunny with a temperature of 75°F."}'
mock_response.json.return_value = {"answer": "The weather today is sunny with a temperature of 75°F."}
mock_response.status_code = 200
mock_client_instance = Mock()
mock_client_instance.request.return_value = mock_response
mock_httpx_client.return_value = mock_client_instance
# Create client with mocked httpx
completion_client = CompletionClient(self.api_key)
response = completion_client.create_completion_message(
def test_create_completion_message(self):
response = self.completion_client.create_completion_message(
{"query": "What's the weather like today?"}, "blocking", "test_user"
)
self.assertIn("answer", response.text)
@patch("dify_client.client.httpx.Client")
def test_create_completion_message_with_vision_model_by_remote_url(self, mock_httpx_client):
# Mock the HTTP response
mock_response = Mock()
mock_response.text = '{"answer": "This is a test image description from completion API."}'
mock_response.json.return_value = {"answer": "This is a test image description from completion API."}
mock_response.status_code = 200
mock_client_instance = Mock()
mock_client_instance.request.return_value = mock_response
mock_httpx_client.return_value = mock_client_instance
# Create client with mocked httpx
completion_client = CompletionClient(self.api_key)
files = [{"type": "image", "transfer_method": "remote_url", "url": "https://example.com/test-image.jpg"}]
response = completion_client.create_completion_message(
def test_create_completion_message_with_vision_model_by_remote_url(self):
files = [{"type": "image", "transfer_method": "remote_url", "url": "your_image_url"}]
response = self.completion_client.create_completion_message(
{"query": "Describe the picture."}, "blocking", "test_user", files
)
self.assertIn("answer", response.text)
@patch("dify_client.client.httpx.Client")
def test_create_completion_message_with_vision_model_by_local_file(self, mock_httpx_client):
# Mock the HTTP response
mock_response = Mock()
mock_response.text = '{"answer": "This is a test uploaded image description from completion API."}'
mock_response.json.return_value = {"answer": "This is a test uploaded image description from completion API."}
mock_response.status_code = 200
mock_client_instance = Mock()
mock_client_instance.request.return_value = mock_response
mock_httpx_client.return_value = mock_client_instance
# Create client with mocked httpx
completion_client = CompletionClient(self.api_key)
def test_create_completion_message_with_vision_model_by_local_file(self):
files = [
{
"type": "image",
"transfer_method": "local_file",
"upload_file_id": "test-file-id",
"upload_file_id": "your_file_id",
}
]
response = completion_client.create_completion_message(
response = self.completion_client.create_completion_message(
{"query": "Describe the picture."}, "blocking", "test_user", files
)
self.assertIn("answer", response.text)
class TestDifyClient(unittest.TestCase):
@patch("dify_client.client.httpx.Client")
def setUp(self, mock_httpx_client):
self.api_key = "test-api-key"
self.dify_client = DifyClient(self.api_key)
def setUp(self):
self.dify_client = DifyClient(API_KEY)
# Set up default mock response for the client
mock_response = Mock()
mock_response.text = '{"result": "success"}'
mock_response.json.return_value = {"result": "success"}
mock_response.status_code = 200
mock_client_instance = Mock()
mock_client_instance.request.return_value = mock_response
mock_httpx_client.return_value = mock_client_instance
@patch("dify_client.client.httpx.Client")
def test_message_feedback(self, mock_httpx_client):
# Mock the HTTP response
mock_response = Mock()
mock_response.text = '{"success": true}'
mock_response.json.return_value = {"success": True}
mock_response.status_code = 200
mock_client_instance = Mock()
mock_client_instance.request.return_value = mock_response
mock_httpx_client.return_value = mock_client_instance
# Create client with mocked httpx
dify_client = DifyClient(self.api_key)
response = dify_client.message_feedback("test-message-id", "like", "test_user")
def test_message_feedback(self):
response = self.dify_client.message_feedback("your_message_id", "like", "test_user")
self.assertIn("success", response.text)
@patch("dify_client.client.httpx.Client")
def test_get_application_parameters(self, mock_httpx_client):
# Mock the HTTP response
mock_response = Mock()
mock_response.text = '{"user_input_form": [{"field": "text", "label": "Input"}]}'
mock_response.json.return_value = {"user_input_form": [{"field": "text", "label": "Input"}]}
mock_response.status_code = 200
mock_client_instance = Mock()
mock_client_instance.request.return_value = mock_response
mock_httpx_client.return_value = mock_client_instance
# Create client with mocked httpx
dify_client = DifyClient(self.api_key)
response = dify_client.get_application_parameters("test_user")
def test_get_application_parameters(self):
response = self.dify_client.get_application_parameters("test_user")
self.assertIn("user_input_form", response.text)
@patch("dify_client.client.httpx.Client")
@patch("builtins.open", new_callable=mock_open, read_data=b"fake image data")
def test_file_upload(self, mock_file_open, mock_httpx_client):
# Mock the HTTP response
mock_response = Mock()
mock_response.text = '{"name": "panda.jpeg", "id": "test-file-id"}'
mock_response.json.return_value = {"name": "panda.jpeg", "id": "test-file-id"}
mock_response.status_code = 200
mock_client_instance = Mock()
mock_client_instance.request.return_value = mock_response
mock_httpx_client.return_value = mock_client_instance
# Create client with mocked httpx
dify_client = DifyClient(self.api_key)
file_path = "/path/to/test/panda.jpeg"
def test_file_upload(self):
file_path = "your_image_file_path"
file_name = "panda.jpeg"
mime_type = "image/jpeg"
with open(file_path, "rb") as file:
files = {"file": (file_name, file, mime_type)}
response = dify_client.file_upload("test_user", files)
response = self.dify_client.file_upload("test_user", files)
self.assertIn("name", response.text)
@@ -1,79 +0,0 @@
"""Tests for custom exceptions."""
import unittest
from dify_client.exceptions import (
DifyClientError,
APIError,
AuthenticationError,
RateLimitError,
ValidationError,
NetworkError,
TimeoutError,
FileUploadError,
DatasetError,
WorkflowError,
)
class TestExceptions(unittest.TestCase):
"""Test custom exception classes."""
def test_base_exception(self):
"""Test base DifyClientError."""
error = DifyClientError("Test message", 500, {"error": "details"})
self.assertEqual(str(error), "Test message")
self.assertEqual(error.status_code, 500)
self.assertEqual(error.response, {"error": "details"})
def test_api_error(self):
"""Test APIError."""
error = APIError("API failed", 400)
self.assertEqual(error.status_code, 400)
self.assertEqual(error.message, "API failed")
def test_authentication_error(self):
"""Test AuthenticationError."""
error = AuthenticationError("Invalid API key")
self.assertEqual(str(error), "Invalid API key")
def test_rate_limit_error(self):
"""Test RateLimitError."""
error = RateLimitError("Rate limited", retry_after=60)
self.assertEqual(error.retry_after, 60)
error_default = RateLimitError()
self.assertEqual(error_default.retry_after, None)
def test_validation_error(self):
"""Test ValidationError."""
error = ValidationError("Invalid parameter")
self.assertEqual(str(error), "Invalid parameter")
def test_network_error(self):
"""Test NetworkError."""
error = NetworkError("Connection failed")
self.assertEqual(str(error), "Connection failed")
def test_timeout_error(self):
"""Test TimeoutError."""
error = TimeoutError("Request timed out")
self.assertEqual(str(error), "Request timed out")
def test_file_upload_error(self):
"""Test FileUploadError."""
error = FileUploadError("Upload failed")
self.assertEqual(str(error), "Upload failed")
def test_dataset_error(self):
"""Test DatasetError."""
error = DatasetError("Dataset operation failed")
self.assertEqual(str(error), "Dataset operation failed")
def test_workflow_error(self):
"""Test WorkflowError."""
error = WorkflowError("Workflow failed")
self.assertEqual(str(error), "Workflow failed")
if __name__ == "__main__":
unittest.main()
@@ -152,7 +152,6 @@ class TestHttpxMigrationMocked(unittest.TestCase):
"""Test that json parameter is passed correctly."""
mock_response = Mock()
mock_response.json.return_value = {"result": "success"}
mock_response.status_code = 200 # Add status_code attribute
mock_client_instance = Mock()
mock_client_instance.request.return_value = mock_response
@@ -174,7 +173,6 @@ class TestHttpxMigrationMocked(unittest.TestCase):
"""Test that params parameter is passed correctly."""
mock_response = Mock()
mock_response.json.return_value = {"result": "success"}
mock_response.status_code = 200 # Add status_code attribute
mock_client_instance = Mock()
mock_client_instance.request.return_value = mock_response
@@ -1,539 +0,0 @@
"""Integration tests with proper mocking."""
import unittest
from unittest.mock import Mock, patch, MagicMock
import json
import httpx
from dify_client import (
DifyClient,
ChatClient,
CompletionClient,
WorkflowClient,
KnowledgeBaseClient,
WorkspaceClient,
)
from dify_client.exceptions import (
APIError,
AuthenticationError,
RateLimitError,
ValidationError,
)
class TestDifyClientIntegration(unittest.TestCase):
"""Integration tests for DifyClient with mocked HTTP responses."""
def setUp(self):
self.api_key = "test_api_key"
self.base_url = "https://api.dify.ai/v1"
self.client = DifyClient(api_key=self.api_key, base_url=self.base_url, enable_logging=False)
@patch("httpx.Client.request")
def test_get_app_info_integration(self, mock_request):
"""Test get_app_info integration."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"id": "app_123",
"name": "Test App",
"description": "A test application",
"mode": "chat",
}
mock_request.return_value = mock_response
response = self.client.get_app_info()
data = response.json()
self.assertEqual(response.status_code, 200)
self.assertEqual(data["id"], "app_123")
self.assertEqual(data["name"], "Test App")
mock_request.assert_called_once_with(
"GET",
"/info",
json=None,
params=None,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
},
)
@patch("httpx.Client.request")
def test_get_application_parameters_integration(self, mock_request):
"""Test get_application_parameters integration."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"opening_statement": "Hello! How can I help you?",
"suggested_questions": ["What is AI?", "How does this work?"],
"speech_to_text": {"enabled": True},
"text_to_speech": {"enabled": False},
}
mock_request.return_value = mock_response
response = self.client.get_application_parameters("user_123")
data = response.json()
self.assertEqual(response.status_code, 200)
self.assertEqual(data["opening_statement"], "Hello! How can I help you?")
self.assertEqual(len(data["suggested_questions"]), 2)
mock_request.assert_called_once_with(
"GET",
"/parameters",
json=None,
params={"user": "user_123"},
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
},
)
@patch("httpx.Client.request")
def test_file_upload_integration(self, mock_request):
"""Test file_upload integration."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"id": "file_123",
"name": "test.txt",
"size": 1024,
"mime_type": "text/plain",
}
mock_request.return_value = mock_response
files = {"file": ("test.txt", "test content", "text/plain")}
response = self.client.file_upload("user_123", files)
data = response.json()
self.assertEqual(response.status_code, 200)
self.assertEqual(data["id"], "file_123")
self.assertEqual(data["name"], "test.txt")
@patch("httpx.Client.request")
def test_message_feedback_integration(self, mock_request):
"""Test message_feedback integration."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {"success": True}
mock_request.return_value = mock_response
response = self.client.message_feedback("msg_123", "like", "user_123")
data = response.json()
self.assertEqual(response.status_code, 200)
self.assertTrue(data["success"])
mock_request.assert_called_once_with(
"POST",
"/messages/msg_123/feedbacks",
json={"rating": "like", "user": "user_123"},
params=None,
headers={
"Authorization": "Bearer test_api_key",
"Content-Type": "application/json",
},
)
class TestChatClientIntegration(unittest.TestCase):
"""Integration tests for ChatClient."""
def setUp(self):
self.client = ChatClient("test_api_key", enable_logging=False)
@patch("httpx.Client.request")
def test_create_chat_message_blocking(self, mock_request):
"""Test create_chat_message with blocking response."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"id": "msg_123",
"answer": "Hello! How can I help you today?",
"conversation_id": "conv_123",
"created_at": 1234567890,
}
mock_request.return_value = mock_response
response = self.client.create_chat_message(
inputs={"query": "Hello"},
query="Hello, AI!",
user="user_123",
response_mode="blocking",
)
data = response.json()
self.assertEqual(response.status_code, 200)
self.assertEqual(data["answer"], "Hello! How can I help you today?")
self.assertEqual(data["conversation_id"], "conv_123")
@patch("httpx.Client.request")
def test_create_chat_message_streaming(self, mock_request):
"""Test create_chat_message with streaming response."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.iter_lines.return_value = [
b'data: {"answer": "Hello"}',
b'data: {"answer": " world"}',
b'data: {"answer": "!"}',
]
mock_request.return_value = mock_response
response = self.client.create_chat_message(inputs={}, query="Hello", user="user_123", response_mode="streaming")
self.assertEqual(response.status_code, 200)
lines = list(response.iter_lines())
self.assertEqual(len(lines), 3)
@patch("httpx.Client.request")
def test_get_conversations_integration(self, mock_request):
"""Test get_conversations integration."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"data": [
{"id": "conv_1", "name": "Conversation 1"},
{"id": "conv_2", "name": "Conversation 2"},
],
"has_more": False,
"limit": 20,
}
mock_request.return_value = mock_response
response = self.client.get_conversations("user_123", limit=20)
data = response.json()
self.assertEqual(response.status_code, 200)
self.assertEqual(len(data["data"]), 2)
self.assertEqual(data["data"][0]["name"], "Conversation 1")
@patch("httpx.Client.request")
def test_get_conversation_messages_integration(self, mock_request):
"""Test get_conversation_messages integration."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"data": [
{"id": "msg_1", "role": "user", "content": "Hello"},
{"id": "msg_2", "role": "assistant", "content": "Hi there!"},
]
}
mock_request.return_value = mock_response
response = self.client.get_conversation_messages("user_123", conversation_id="conv_123")
data = response.json()
self.assertEqual(response.status_code, 200)
self.assertEqual(len(data["data"]), 2)
self.assertEqual(data["data"][0]["role"], "user")
class TestCompletionClientIntegration(unittest.TestCase):
"""Integration tests for CompletionClient."""
def setUp(self):
self.client = CompletionClient("test_api_key", enable_logging=False)
@patch("httpx.Client.request")
def test_create_completion_message_blocking(self, mock_request):
"""Test create_completion_message with blocking response."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"id": "comp_123",
"answer": "This is a completion response.",
"created_at": 1234567890,
}
mock_request.return_value = mock_response
response = self.client.create_completion_message(
inputs={"prompt": "Complete this sentence"},
response_mode="blocking",
user="user_123",
)
data = response.json()
self.assertEqual(response.status_code, 200)
self.assertEqual(data["answer"], "This is a completion response.")
@patch("httpx.Client.request")
def test_create_completion_message_with_files(self, mock_request):
"""Test create_completion_message with files."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"id": "comp_124",
"answer": "I can see the image shows...",
"files": [{"id": "file_1", "type": "image"}],
}
mock_request.return_value = mock_response
files = {
"file": {
"type": "image",
"transfer_method": "remote_url",
"url": "https://example.com/image.jpg",
}
}
response = self.client.create_completion_message(
inputs={"prompt": "Describe this image"},
response_mode="blocking",
user="user_123",
files=files,
)
data = response.json()
self.assertEqual(response.status_code, 200)
self.assertIn("image", data["answer"])
self.assertEqual(len(data["files"]), 1)
class TestWorkflowClientIntegration(unittest.TestCase):
"""Integration tests for WorkflowClient."""
def setUp(self):
self.client = WorkflowClient("test_api_key", enable_logging=False)
@patch("httpx.Client.request")
def test_run_workflow_blocking(self, mock_request):
"""Test run workflow with blocking response."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"id": "run_123",
"workflow_id": "workflow_123",
"status": "succeeded",
"inputs": {"query": "Test input"},
"outputs": {"result": "Test output"},
"elapsed_time": 2.5,
}
mock_request.return_value = mock_response
response = self.client.run(inputs={"query": "Test input"}, response_mode="blocking", user="user_123")
data = response.json()
self.assertEqual(response.status_code, 200)
self.assertEqual(data["status"], "succeeded")
self.assertEqual(data["outputs"]["result"], "Test output")
@patch("httpx.Client.request")
def test_get_workflow_logs(self, mock_request):
"""Test get_workflow_logs integration."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"logs": [
{"id": "log_1", "status": "succeeded", "created_at": 1234567890},
{"id": "log_2", "status": "failed", "created_at": 1234567891},
],
"total": 2,
"page": 1,
"limit": 20,
}
mock_request.return_value = mock_response
response = self.client.get_workflow_logs(page=1, limit=20)
data = response.json()
self.assertEqual(response.status_code, 200)
self.assertEqual(len(data["logs"]), 2)
self.assertEqual(data["logs"][0]["status"], "succeeded")
class TestKnowledgeBaseClientIntegration(unittest.TestCase):
"""Integration tests for KnowledgeBaseClient."""
def setUp(self):
self.client = KnowledgeBaseClient("test_api_key")
@patch("httpx.Client.request")
def test_create_dataset(self, mock_request):
"""Test create_dataset integration."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"id": "dataset_123",
"name": "Test Dataset",
"description": "A test dataset",
"created_at": 1234567890,
}
mock_request.return_value = mock_response
response = self.client.create_dataset(name="Test Dataset")
data = response.json()
self.assertEqual(response.status_code, 200)
self.assertEqual(data["name"], "Test Dataset")
self.assertEqual(data["id"], "dataset_123")
@patch("httpx.Client.request")
def test_list_datasets(self, mock_request):
"""Test list_datasets integration."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"data": [
{"id": "dataset_1", "name": "Dataset 1"},
{"id": "dataset_2", "name": "Dataset 2"},
],
"has_more": False,
"limit": 20,
}
mock_request.return_value = mock_response
response = self.client.list_datasets(page=1, page_size=20)
data = response.json()
self.assertEqual(response.status_code, 200)
self.assertEqual(len(data["data"]), 2)
@patch("httpx.Client.request")
def test_create_document_by_text(self, mock_request):
"""Test create_document_by_text integration."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"document": {
"id": "doc_123",
"name": "Test Document",
"word_count": 100,
"status": "indexing",
}
}
mock_request.return_value = mock_response
# Mock dataset_id
self.client.dataset_id = "dataset_123"
response = self.client.create_document_by_text(name="Test Document", text="This is test document content.")
data = response.json()
self.assertEqual(response.status_code, 200)
self.assertEqual(data["document"]["name"], "Test Document")
self.assertEqual(data["document"]["word_count"], 100)
class TestWorkspaceClientIntegration(unittest.TestCase):
"""Integration tests for WorkspaceClient."""
def setUp(self):
self.client = WorkspaceClient("test_api_key", enable_logging=False)
@patch("httpx.Client.request")
def test_get_available_models(self, mock_request):
"""Test get_available_models integration."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"models": [
{"id": "gpt-4", "name": "GPT-4", "provider": "openai"},
{"id": "claude-3", "name": "Claude 3", "provider": "anthropic"},
]
}
mock_request.return_value = mock_response
response = self.client.get_available_models("llm")
data = response.json()
self.assertEqual(response.status_code, 200)
self.assertEqual(len(data["models"]), 2)
self.assertEqual(data["models"][0]["id"], "gpt-4")
class TestErrorScenariosIntegration(unittest.TestCase):
"""Integration tests for error scenarios."""
def setUp(self):
self.client = DifyClient("test_api_key", enable_logging=False)
@patch("httpx.Client.request")
def test_authentication_error_integration(self, mock_request):
"""Test authentication error in integration."""
mock_response = Mock()
mock_response.status_code = 401
mock_response.json.return_value = {"message": "Invalid API key"}
mock_request.return_value = mock_response
with self.assertRaises(AuthenticationError) as context:
self.client.get_app_info()
self.assertEqual(str(context.exception), "Invalid API key")
self.assertEqual(context.exception.status_code, 401)
@patch("httpx.Client.request")
def test_rate_limit_error_integration(self, mock_request):
"""Test rate limit error in integration."""
mock_response = Mock()
mock_response.status_code = 429
mock_response.json.return_value = {"message": "Rate limit exceeded"}
mock_response.headers = {"Retry-After": "60"}
mock_request.return_value = mock_response
with self.assertRaises(RateLimitError) as context:
self.client.get_app_info()
self.assertEqual(str(context.exception), "Rate limit exceeded")
self.assertEqual(context.exception.retry_after, "60")
@patch("httpx.Client.request")
def test_server_error_with_retry_integration(self, mock_request):
"""Test server error with retry in integration."""
# API errors don't retry by design - only network/timeout errors retry
mock_response_500 = Mock()
mock_response_500.status_code = 500
mock_response_500.json.return_value = {"message": "Internal server error"}
mock_request.return_value = mock_response_500
with patch("time.sleep"): # Skip actual sleep
with self.assertRaises(APIError) as context:
self.client.get_app_info()
self.assertEqual(str(context.exception), "Internal server error")
self.assertEqual(mock_request.call_count, 1)
@patch("httpx.Client.request")
def test_validation_error_integration(self, mock_request):
"""Test validation error in integration."""
mock_response = Mock()
mock_response.status_code = 422
mock_response.json.return_value = {
"message": "Validation failed",
"details": {"field": "query", "error": "required"},
}
mock_request.return_value = mock_response
with self.assertRaises(ValidationError) as context:
self.client.get_app_info()
self.assertEqual(str(context.exception), "Validation failed")
self.assertEqual(context.exception.status_code, 422)
class TestContextManagerIntegration(unittest.TestCase):
"""Integration tests for context manager usage."""
@patch("httpx.Client.close")
@patch("httpx.Client.request")
def test_context_manager_usage(self, mock_request, mock_close):
"""Test context manager properly closes connections."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {"id": "app_123", "name": "Test App"}
mock_request.return_value = mock_response
with DifyClient("test_api_key") as client:
response = client.get_app_info()
self.assertEqual(response.status_code, 200)
# Verify close was called
mock_close.assert_called_once()
@patch("httpx.Client.close")
def test_manual_close(self, mock_close):
"""Test manual close method."""
client = DifyClient("test_api_key")
client.close()
mock_close.assert_called_once()
if __name__ == "__main__":
unittest.main()
-640
View File
@@ -1,640 +0,0 @@
"""Unit tests for response models."""
import unittest
import json
from datetime import datetime
from dify_client.models import (
BaseResponse,
ErrorResponse,
FileInfo,
MessageResponse,
ConversationResponse,
DatasetResponse,
DocumentResponse,
DocumentSegmentResponse,
WorkflowRunResponse,
ApplicationParametersResponse,
AnnotationResponse,
PaginatedResponse,
ConversationVariableResponse,
FileUploadResponse,
AudioResponse,
SuggestedQuestionsResponse,
AppInfoResponse,
WorkspaceModelsResponse,
HitTestingResponse,
DatasetTagsResponse,
WorkflowLogsResponse,
ModelProviderResponse,
FileInfoResponse,
WorkflowDraftResponse,
ApiTokenResponse,
JobStatusResponse,
DatasetQueryResponse,
DatasetTemplateResponse,
)
class TestResponseModels(unittest.TestCase):
"""Test cases for response model classes."""
def test_base_response(self):
"""Test BaseResponse model."""
response = BaseResponse(success=True, message="Operation successful")
self.assertTrue(response.success)
self.assertEqual(response.message, "Operation successful")
def test_base_response_defaults(self):
"""Test BaseResponse with default values."""
response = BaseResponse(success=True)
self.assertTrue(response.success)
self.assertIsNone(response.message)
def test_error_response(self):
"""Test ErrorResponse model."""
response = ErrorResponse(
success=False,
message="Error occurred",
error_code="VALIDATION_ERROR",
details={"field": "invalid_value"},
)
self.assertFalse(response.success)
self.assertEqual(response.message, "Error occurred")
self.assertEqual(response.error_code, "VALIDATION_ERROR")
self.assertEqual(response.details["field"], "invalid_value")
def test_file_info(self):
"""Test FileInfo model."""
now = datetime.now()
file_info = FileInfo(
id="file_123",
name="test.txt",
size=1024,
mime_type="text/plain",
url="https://example.com/file.txt",
created_at=now,
)
self.assertEqual(file_info.id, "file_123")
self.assertEqual(file_info.name, "test.txt")
self.assertEqual(file_info.size, 1024)
self.assertEqual(file_info.mime_type, "text/plain")
self.assertEqual(file_info.url, "https://example.com/file.txt")
self.assertEqual(file_info.created_at, now)
def test_message_response(self):
"""Test MessageResponse model."""
response = MessageResponse(
success=True,
id="msg_123",
answer="Hello, world!",
conversation_id="conv_123",
created_at=1234567890,
metadata={"model": "gpt-4"},
files=[{"id": "file_1", "type": "image"}],
)
self.assertTrue(response.success)
self.assertEqual(response.id, "msg_123")
self.assertEqual(response.answer, "Hello, world!")
self.assertEqual(response.conversation_id, "conv_123")
self.assertEqual(response.created_at, 1234567890)
self.assertEqual(response.metadata["model"], "gpt-4")
self.assertEqual(response.files[0]["id"], "file_1")
def test_conversation_response(self):
"""Test ConversationResponse model."""
response = ConversationResponse(
success=True,
id="conv_123",
name="Test Conversation",
inputs={"query": "Hello"},
status="active",
created_at=1234567890,
updated_at=1234567891,
)
self.assertTrue(response.success)
self.assertEqual(response.id, "conv_123")
self.assertEqual(response.name, "Test Conversation")
self.assertEqual(response.inputs["query"], "Hello")
self.assertEqual(response.status, "active")
self.assertEqual(response.created_at, 1234567890)
self.assertEqual(response.updated_at, 1234567891)
def test_dataset_response(self):
"""Test DatasetResponse model."""
response = DatasetResponse(
success=True,
id="dataset_123",
name="Test Dataset",
description="A test dataset",
permission="read",
indexing_technique="high_quality",
embedding_model="text-embedding-ada-002",
embedding_model_provider="openai",
retrieval_model={"search_type": "semantic"},
document_count=10,
word_count=5000,
app_count=2,
created_at=1234567890,
updated_at=1234567891,
)
self.assertTrue(response.success)
self.assertEqual(response.id, "dataset_123")
self.assertEqual(response.name, "Test Dataset")
self.assertEqual(response.description, "A test dataset")
self.assertEqual(response.permission, "read")
self.assertEqual(response.indexing_technique, "high_quality")
self.assertEqual(response.embedding_model, "text-embedding-ada-002")
self.assertEqual(response.embedding_model_provider, "openai")
self.assertEqual(response.retrieval_model["search_type"], "semantic")
self.assertEqual(response.document_count, 10)
self.assertEqual(response.word_count, 5000)
self.assertEqual(response.app_count, 2)
def test_document_response(self):
"""Test DocumentResponse model."""
response = DocumentResponse(
success=True,
id="doc_123",
name="test_document.txt",
data_source_type="upload_file",
position=1,
enabled=True,
word_count=1000,
hit_count=5,
doc_form="text_model",
created_at=1234567890.0,
indexing_status="completed",
completed_at=1234567891.0,
)
self.assertTrue(response.success)
self.assertEqual(response.id, "doc_123")
self.assertEqual(response.name, "test_document.txt")
self.assertEqual(response.data_source_type, "upload_file")
self.assertEqual(response.position, 1)
self.assertTrue(response.enabled)
self.assertEqual(response.word_count, 1000)
self.assertEqual(response.hit_count, 5)
self.assertEqual(response.doc_form, "text_model")
self.assertEqual(response.created_at, 1234567890.0)
self.assertEqual(response.indexing_status, "completed")
self.assertEqual(response.completed_at, 1234567891.0)
def test_document_segment_response(self):
"""Test DocumentSegmentResponse model."""
response = DocumentSegmentResponse(
success=True,
id="seg_123",
position=1,
document_id="doc_123",
content="This is a test segment.",
answer="Test answer",
word_count=5,
tokens=10,
keywords=["test", "segment"],
hit_count=2,
enabled=True,
status="completed",
created_at=1234567890.0,
completed_at=1234567891.0,
)
self.assertTrue(response.success)
self.assertEqual(response.id, "seg_123")
self.assertEqual(response.position, 1)
self.assertEqual(response.document_id, "doc_123")
self.assertEqual(response.content, "This is a test segment.")
self.assertEqual(response.answer, "Test answer")
self.assertEqual(response.word_count, 5)
self.assertEqual(response.tokens, 10)
self.assertEqual(response.keywords, ["test", "segment"])
self.assertEqual(response.hit_count, 2)
self.assertTrue(response.enabled)
self.assertEqual(response.status, "completed")
self.assertEqual(response.created_at, 1234567890.0)
self.assertEqual(response.completed_at, 1234567891.0)
def test_workflow_run_response(self):
"""Test WorkflowRunResponse model."""
response = WorkflowRunResponse(
success=True,
id="run_123",
workflow_id="workflow_123",
status="succeeded",
inputs={"query": "test"},
outputs={"answer": "result"},
elapsed_time=5.5,
total_tokens=100,
total_steps=3,
created_at=1234567890.0,
finished_at=1234567895.5,
)
self.assertTrue(response.success)
self.assertEqual(response.id, "run_123")
self.assertEqual(response.workflow_id, "workflow_123")
self.assertEqual(response.status, "succeeded")
self.assertEqual(response.inputs["query"], "test")
self.assertEqual(response.outputs["answer"], "result")
self.assertEqual(response.elapsed_time, 5.5)
self.assertEqual(response.total_tokens, 100)
self.assertEqual(response.total_steps, 3)
self.assertEqual(response.created_at, 1234567890.0)
self.assertEqual(response.finished_at, 1234567895.5)
def test_application_parameters_response(self):
"""Test ApplicationParametersResponse model."""
response = ApplicationParametersResponse(
success=True,
opening_statement="Hello! How can I help you?",
suggested_questions=["What is AI?", "How does this work?"],
speech_to_text={"enabled": True},
text_to_speech={"enabled": False, "voice": "alloy"},
retriever_resource={"enabled": True},
sensitive_word_avoidance={"enabled": False},
file_upload={"enabled": True, "file_size_limit": 10485760},
system_parameters={"max_tokens": 1000},
user_input_form=[{"type": "text", "label": "Query"}],
)
self.assertTrue(response.success)
self.assertEqual(response.opening_statement, "Hello! How can I help you?")
self.assertEqual(response.suggested_questions, ["What is AI?", "How does this work?"])
self.assertTrue(response.speech_to_text["enabled"])
self.assertFalse(response.text_to_speech["enabled"])
self.assertEqual(response.text_to_speech["voice"], "alloy")
self.assertTrue(response.retriever_resource["enabled"])
self.assertFalse(response.sensitive_word_avoidance["enabled"])
self.assertTrue(response.file_upload["enabled"])
self.assertEqual(response.file_upload["file_size_limit"], 10485760)
self.assertEqual(response.system_parameters["max_tokens"], 1000)
self.assertEqual(response.user_input_form[0]["type"], "text")
def test_annotation_response(self):
"""Test AnnotationResponse model."""
response = AnnotationResponse(
success=True,
id="annotation_123",
question="What is the capital of France?",
answer="Paris",
content="Additional context",
created_at=1234567890.0,
updated_at=1234567891.0,
created_by="user_123",
updated_by="user_123",
hit_count=5,
)
self.assertTrue(response.success)
self.assertEqual(response.id, "annotation_123")
self.assertEqual(response.question, "What is the capital of France?")
self.assertEqual(response.answer, "Paris")
self.assertEqual(response.content, "Additional context")
self.assertEqual(response.created_at, 1234567890.0)
self.assertEqual(response.updated_at, 1234567891.0)
self.assertEqual(response.created_by, "user_123")
self.assertEqual(response.updated_by, "user_123")
self.assertEqual(response.hit_count, 5)
def test_paginated_response(self):
"""Test PaginatedResponse model."""
response = PaginatedResponse(
success=True,
data=[{"id": 1}, {"id": 2}, {"id": 3}],
has_more=True,
limit=10,
total=100,
page=1,
)
self.assertTrue(response.success)
self.assertEqual(len(response.data), 3)
self.assertEqual(response.data[0]["id"], 1)
self.assertTrue(response.has_more)
self.assertEqual(response.limit, 10)
self.assertEqual(response.total, 100)
self.assertEqual(response.page, 1)
def test_conversation_variable_response(self):
"""Test ConversationVariableResponse model."""
response = ConversationVariableResponse(
success=True,
conversation_id="conv_123",
variables=[
{"id": "var_1", "name": "user_name", "value": "John"},
{"id": "var_2", "name": "preferences", "value": {"theme": "dark"}},
],
)
self.assertTrue(response.success)
self.assertEqual(response.conversation_id, "conv_123")
self.assertEqual(len(response.variables), 2)
self.assertEqual(response.variables[0]["name"], "user_name")
self.assertEqual(response.variables[0]["value"], "John")
self.assertEqual(response.variables[1]["name"], "preferences")
self.assertEqual(response.variables[1]["value"]["theme"], "dark")
def test_file_upload_response(self):
"""Test FileUploadResponse model."""
response = FileUploadResponse(
success=True,
id="file_123",
name="test.txt",
size=1024,
mime_type="text/plain",
url="https://example.com/files/test.txt",
created_at=1234567890.0,
)
self.assertTrue(response.success)
self.assertEqual(response.id, "file_123")
self.assertEqual(response.name, "test.txt")
self.assertEqual(response.size, 1024)
self.assertEqual(response.mime_type, "text/plain")
self.assertEqual(response.url, "https://example.com/files/test.txt")
self.assertEqual(response.created_at, 1234567890.0)
def test_audio_response(self):
"""Test AudioResponse model."""
response = AudioResponse(
success=True,
audio="base64_encoded_audio_data",
audio_url="https://example.com/audio.mp3",
duration=10.5,
sample_rate=44100,
)
self.assertTrue(response.success)
self.assertEqual(response.audio, "base64_encoded_audio_data")
self.assertEqual(response.audio_url, "https://example.com/audio.mp3")
self.assertEqual(response.duration, 10.5)
self.assertEqual(response.sample_rate, 44100)
def test_suggested_questions_response(self):
"""Test SuggestedQuestionsResponse model."""
response = SuggestedQuestionsResponse(
success=True,
message_id="msg_123",
questions=[
"What is machine learning?",
"How does AI work?",
"Can you explain neural networks?",
],
)
self.assertTrue(response.success)
self.assertEqual(response.message_id, "msg_123")
self.assertEqual(len(response.questions), 3)
self.assertEqual(response.questions[0], "What is machine learning?")
def test_app_info_response(self):
"""Test AppInfoResponse model."""
response = AppInfoResponse(
success=True,
id="app_123",
name="Test App",
description="A test application",
icon="🤖",
icon_background="#FF6B6B",
mode="chat",
tags=["AI", "Chat", "Test"],
enable_site=True,
enable_api=True,
api_token="app_token_123",
)
self.assertTrue(response.success)
self.assertEqual(response.id, "app_123")
self.assertEqual(response.name, "Test App")
self.assertEqual(response.description, "A test application")
self.assertEqual(response.icon, "🤖")
self.assertEqual(response.icon_background, "#FF6B6B")
self.assertEqual(response.mode, "chat")
self.assertEqual(response.tags, ["AI", "Chat", "Test"])
self.assertTrue(response.enable_site)
self.assertTrue(response.enable_api)
self.assertEqual(response.api_token, "app_token_123")
def test_workspace_models_response(self):
"""Test WorkspaceModelsResponse model."""
response = WorkspaceModelsResponse(
success=True,
models=[
{"id": "gpt-4", "name": "GPT-4", "provider": "openai"},
{"id": "claude-3", "name": "Claude 3", "provider": "anthropic"},
],
)
self.assertTrue(response.success)
self.assertEqual(len(response.models), 2)
self.assertEqual(response.models[0]["id"], "gpt-4")
self.assertEqual(response.models[0]["name"], "GPT-4")
self.assertEqual(response.models[0]["provider"], "openai")
def test_hit_testing_response(self):
"""Test HitTestingResponse model."""
response = HitTestingResponse(
success=True,
query="What is machine learning?",
records=[
{"content": "Machine learning is a subset of AI...", "score": 0.95},
{"content": "ML algorithms learn from data...", "score": 0.87},
],
)
self.assertTrue(response.success)
self.assertEqual(response.query, "What is machine learning?")
self.assertEqual(len(response.records), 2)
self.assertEqual(response.records[0]["score"], 0.95)
def test_dataset_tags_response(self):
"""Test DatasetTagsResponse model."""
response = DatasetTagsResponse(
success=True,
tags=[
{"id": "tag_1", "name": "Technology", "color": "#FF0000"},
{"id": "tag_2", "name": "Science", "color": "#00FF00"},
],
)
self.assertTrue(response.success)
self.assertEqual(len(response.tags), 2)
self.assertEqual(response.tags[0]["name"], "Technology")
self.assertEqual(response.tags[0]["color"], "#FF0000")
def test_workflow_logs_response(self):
"""Test WorkflowLogsResponse model."""
response = WorkflowLogsResponse(
success=True,
logs=[
{"id": "log_1", "status": "succeeded", "created_at": 1234567890},
{"id": "log_2", "status": "failed", "created_at": 1234567891},
],
total=50,
page=1,
limit=10,
has_more=True,
)
self.assertTrue(response.success)
self.assertEqual(len(response.logs), 2)
self.assertEqual(response.logs[0]["status"], "succeeded")
self.assertEqual(response.total, 50)
self.assertEqual(response.page, 1)
self.assertEqual(response.limit, 10)
self.assertTrue(response.has_more)
def test_model_serialization(self):
"""Test that models can be serialized to JSON."""
response = MessageResponse(
success=True,
id="msg_123",
answer="Hello, world!",
conversation_id="conv_123",
)
# Convert to dict and then to JSON
response_dict = {
"success": response.success,
"id": response.id,
"answer": response.answer,
"conversation_id": response.conversation_id,
}
json_str = json.dumps(response_dict)
parsed = json.loads(json_str)
self.assertTrue(parsed["success"])
self.assertEqual(parsed["id"], "msg_123")
self.assertEqual(parsed["answer"], "Hello, world!")
self.assertEqual(parsed["conversation_id"], "conv_123")
# Tests for new response models
def test_model_provider_response(self):
"""Test ModelProviderResponse model."""
response = ModelProviderResponse(
success=True,
provider_name="openai",
provider_type="llm",
models=[
{"id": "gpt-4", "name": "GPT-4", "max_tokens": 8192},
{"id": "gpt-3.5-turbo", "name": "GPT-3.5 Turbo", "max_tokens": 4096},
],
is_enabled=True,
credentials={"api_key": "sk-..."},
)
self.assertTrue(response.success)
self.assertEqual(response.provider_name, "openai")
self.assertEqual(response.provider_type, "llm")
self.assertEqual(len(response.models), 2)
self.assertEqual(response.models[0]["id"], "gpt-4")
self.assertTrue(response.is_enabled)
self.assertEqual(response.credentials["api_key"], "sk-...")
def test_file_info_response(self):
"""Test FileInfoResponse model."""
response = FileInfoResponse(
success=True,
id="file_123",
name="document.pdf",
size=2048576,
mime_type="application/pdf",
url="https://example.com/files/document.pdf",
created_at=1234567890,
metadata={"pages": 10, "author": "John Doe"},
)
self.assertTrue(response.success)
self.assertEqual(response.id, "file_123")
self.assertEqual(response.name, "document.pdf")
self.assertEqual(response.size, 2048576)
self.assertEqual(response.mime_type, "application/pdf")
self.assertEqual(response.url, "https://example.com/files/document.pdf")
self.assertEqual(response.created_at, 1234567890)
self.assertEqual(response.metadata["pages"], 10)
def test_workflow_draft_response(self):
"""Test WorkflowDraftResponse model."""
response = WorkflowDraftResponse(
success=True,
id="draft_123",
app_id="app_456",
draft_data={"nodes": [], "edges": [], "config": {"name": "Test Workflow"}},
version=1,
created_at=1234567890,
updated_at=1234567891,
)
self.assertTrue(response.success)
self.assertEqual(response.id, "draft_123")
self.assertEqual(response.app_id, "app_456")
self.assertEqual(response.draft_data["config"]["name"], "Test Workflow")
self.assertEqual(response.version, 1)
self.assertEqual(response.created_at, 1234567890)
self.assertEqual(response.updated_at, 1234567891)
def test_api_token_response(self):
"""Test ApiTokenResponse model."""
response = ApiTokenResponse(
success=True,
id="token_123",
name="Production Token",
token="app-xxxxxxxxxxxx",
description="Token for production environment",
created_at=1234567890,
last_used_at=1234567891,
is_active=True,
)
self.assertTrue(response.success)
self.assertEqual(response.id, "token_123")
self.assertEqual(response.name, "Production Token")
self.assertEqual(response.token, "app-xxxxxxxxxxxx")
self.assertEqual(response.description, "Token for production environment")
self.assertEqual(response.created_at, 1234567890)
self.assertEqual(response.last_used_at, 1234567891)
self.assertTrue(response.is_active)
def test_job_status_response(self):
"""Test JobStatusResponse model."""
response = JobStatusResponse(
success=True,
job_id="job_123",
job_status="running",
error_msg=None,
progress=0.75,
created_at=1234567890,
updated_at=1234567891,
)
self.assertTrue(response.success)
self.assertEqual(response.job_id, "job_123")
self.assertEqual(response.job_status, "running")
self.assertIsNone(response.error_msg)
self.assertEqual(response.progress, 0.75)
self.assertEqual(response.created_at, 1234567890)
self.assertEqual(response.updated_at, 1234567891)
def test_dataset_query_response(self):
"""Test DatasetQueryResponse model."""
response = DatasetQueryResponse(
success=True,
query="What is machine learning?",
records=[
{"content": "Machine learning is...", "score": 0.95},
{"content": "ML algorithms...", "score": 0.87},
],
total=2,
search_time=0.123,
retrieval_model={"method": "semantic_search", "top_k": 3},
)
self.assertTrue(response.success)
self.assertEqual(response.query, "What is machine learning?")
self.assertEqual(len(response.records), 2)
self.assertEqual(response.total, 2)
self.assertEqual(response.search_time, 0.123)
self.assertEqual(response.retrieval_model["method"], "semantic_search")
def test_dataset_template_response(self):
"""Test DatasetTemplateResponse model."""
response = DatasetTemplateResponse(
success=True,
template_name="customer_support",
display_name="Customer Support",
description="Template for customer support knowledge base",
category="support",
icon="🎧",
config_schema={"fields": [{"name": "category", "type": "string"}]},
)
self.assertTrue(response.success)
self.assertEqual(response.template_name, "customer_support")
self.assertEqual(response.display_name, "Customer Support")
self.assertEqual(response.description, "Template for customer support knowledge base")
self.assertEqual(response.category, "support")
self.assertEqual(response.icon, "🎧")
self.assertEqual(response.config_schema["fields"][0]["name"], "category")
if __name__ == "__main__":
unittest.main()
@@ -1,313 +0,0 @@
"""Unit tests for retry mechanism and error handling."""
import unittest
from unittest.mock import Mock, patch, MagicMock
import httpx
from dify_client.client import DifyClient
from dify_client.exceptions import (
APIError,
AuthenticationError,
RateLimitError,
ValidationError,
NetworkError,
TimeoutError,
FileUploadError,
)
class TestRetryMechanism(unittest.TestCase):
"""Test cases for retry mechanism."""
def setUp(self):
self.api_key = "test_api_key"
self.base_url = "https://api.dify.ai/v1"
self.client = DifyClient(
api_key=self.api_key,
base_url=self.base_url,
max_retries=3,
retry_delay=0.1, # Short delay for tests
enable_logging=False,
)
@patch("httpx.Client.request")
def test_successful_request_no_retry(self, mock_request):
"""Test that successful requests don't trigger retries."""
mock_response = Mock()
mock_response.status_code = 200
mock_response.content = b'{"success": true}'
mock_request.return_value = mock_response
response = self.client._send_request("GET", "/test")
self.assertEqual(response, mock_response)
self.assertEqual(mock_request.call_count, 1)
@patch("httpx.Client.request")
@patch("time.sleep")
def test_retry_on_network_error(self, mock_sleep, mock_request):
"""Test retry on network errors."""
# First two calls raise network error, third succeeds
mock_request.side_effect = [
httpx.NetworkError("Connection failed"),
httpx.NetworkError("Connection failed"),
Mock(status_code=200, content=b'{"success": true}'),
]
mock_response = Mock()
mock_response.status_code = 200
mock_response.content = b'{"success": true}'
response = self.client._send_request("GET", "/test")
self.assertEqual(response.status_code, 200)
self.assertEqual(mock_request.call_count, 3)
self.assertEqual(mock_sleep.call_count, 2)
@patch("httpx.Client.request")
@patch("time.sleep")
def test_retry_on_timeout_error(self, mock_sleep, mock_request):
"""Test retry on timeout errors."""
mock_request.side_effect = [
httpx.TimeoutException("Request timed out"),
httpx.TimeoutException("Request timed out"),
Mock(status_code=200, content=b'{"success": true}'),
]
response = self.client._send_request("GET", "/test")
self.assertEqual(response.status_code, 200)
self.assertEqual(mock_request.call_count, 3)
self.assertEqual(mock_sleep.call_count, 2)
@patch("httpx.Client.request")
@patch("time.sleep")
def test_max_retries_exceeded(self, mock_sleep, mock_request):
"""Test behavior when max retries are exceeded."""
mock_request.side_effect = httpx.NetworkError("Persistent network error")
with self.assertRaises(NetworkError):
self.client._send_request("GET", "/test")
self.assertEqual(mock_request.call_count, 4) # 1 initial + 3 retries
self.assertEqual(mock_sleep.call_count, 3)
@patch("httpx.Client.request")
def test_no_retry_on_client_error(self, mock_request):
"""Test that client errors (4xx) don't trigger retries."""
mock_response = Mock()
mock_response.status_code = 401
mock_response.json.return_value = {"message": "Unauthorized"}
mock_request.return_value = mock_response
with self.assertRaises(AuthenticationError):
self.client._send_request("GET", "/test")
self.assertEqual(mock_request.call_count, 1)
@patch("httpx.Client.request")
def test_retry_on_server_error(self, mock_request):
"""Test that server errors (5xx) don't retry - they raise APIError immediately."""
mock_response_500 = Mock()
mock_response_500.status_code = 500
mock_response_500.json.return_value = {"message": "Internal server error"}
mock_request.return_value = mock_response_500
with self.assertRaises(APIError) as context:
self.client._send_request("GET", "/test")
self.assertEqual(str(context.exception), "Internal server error")
self.assertEqual(context.exception.status_code, 500)
# Should not retry server errors
self.assertEqual(mock_request.call_count, 1)
@patch("httpx.Client.request")
def test_exponential_backoff(self, mock_request):
"""Test exponential backoff timing."""
mock_request.side_effect = [
httpx.NetworkError("Connection failed"),
httpx.NetworkError("Connection failed"),
httpx.NetworkError("Connection failed"),
httpx.NetworkError("Connection failed"), # All attempts fail
]
with patch("time.sleep") as mock_sleep:
with self.assertRaises(NetworkError):
self.client._send_request("GET", "/test")
# Check exponential backoff: 0.1, 0.2, 0.4
expected_calls = [0.1, 0.2, 0.4]
actual_calls = [call[0][0] for call in mock_sleep.call_args_list]
self.assertEqual(actual_calls, expected_calls)
class TestErrorHandling(unittest.TestCase):
"""Test cases for error handling."""
def setUp(self):
self.client = DifyClient(api_key="test_api_key", enable_logging=False)
@patch("httpx.Client.request")
def test_authentication_error(self, mock_request):
"""Test AuthenticationError handling."""
mock_response = Mock()
mock_response.status_code = 401
mock_response.json.return_value = {"message": "Invalid API key"}
mock_request.return_value = mock_response
with self.assertRaises(AuthenticationError) as context:
self.client._send_request("GET", "/test")
self.assertEqual(str(context.exception), "Invalid API key")
self.assertEqual(context.exception.status_code, 401)
@patch("httpx.Client.request")
def test_rate_limit_error(self, mock_request):
"""Test RateLimitError handling."""
mock_response = Mock()
mock_response.status_code = 429
mock_response.json.return_value = {"message": "Rate limit exceeded"}
mock_response.headers = {"Retry-After": "60"}
mock_request.return_value = mock_response
with self.assertRaises(RateLimitError) as context:
self.client._send_request("GET", "/test")
self.assertEqual(str(context.exception), "Rate limit exceeded")
self.assertEqual(context.exception.retry_after, "60")
@patch("httpx.Client.request")
def test_validation_error(self, mock_request):
"""Test ValidationError handling."""
mock_response = Mock()
mock_response.status_code = 422
mock_response.json.return_value = {"message": "Invalid parameters"}
mock_request.return_value = mock_response
with self.assertRaises(ValidationError) as context:
self.client._send_request("GET", "/test")
self.assertEqual(str(context.exception), "Invalid parameters")
self.assertEqual(context.exception.status_code, 422)
@patch("httpx.Client.request")
def test_api_error(self, mock_request):
"""Test general APIError handling."""
mock_response = Mock()
mock_response.status_code = 500
mock_response.json.return_value = {"message": "Internal server error"}
mock_request.return_value = mock_response
with self.assertRaises(APIError) as context:
self.client._send_request("GET", "/test")
self.assertEqual(str(context.exception), "Internal server error")
self.assertEqual(context.exception.status_code, 500)
@patch("httpx.Client.request")
def test_error_response_without_json(self, mock_request):
"""Test error handling when response doesn't contain valid JSON."""
mock_response = Mock()
mock_response.status_code = 500
mock_response.content = b"Internal Server Error"
mock_response.json.side_effect = ValueError("No JSON object could be decoded")
mock_request.return_value = mock_response
with self.assertRaises(APIError) as context:
self.client._send_request("GET", "/test")
self.assertEqual(str(context.exception), "HTTP 500")
@patch("httpx.Client.request")
def test_file_upload_error(self, mock_request):
"""Test FileUploadError handling."""
mock_response = Mock()
mock_response.status_code = 400
mock_response.json.return_value = {"message": "File upload failed"}
mock_request.return_value = mock_response
with self.assertRaises(FileUploadError) as context:
self.client._send_request_with_files("POST", "/upload", {}, {})
self.assertEqual(str(context.exception), "File upload failed")
self.assertEqual(context.exception.status_code, 400)
class TestParameterValidation(unittest.TestCase):
"""Test cases for parameter validation."""
def setUp(self):
self.client = DifyClient(api_key="test_api_key", enable_logging=False)
def test_empty_string_validation(self):
"""Test validation of empty strings."""
with self.assertRaises(ValidationError):
self.client._validate_params(empty_string="")
def test_whitespace_only_string_validation(self):
"""Test validation of whitespace-only strings."""
with self.assertRaises(ValidationError):
self.client._validate_params(whitespace_string=" ")
def test_long_string_validation(self):
"""Test validation of overly long strings."""
long_string = "a" * 10001 # Exceeds 10000 character limit
with self.assertRaises(ValidationError):
self.client._validate_params(long_string=long_string)
def test_large_list_validation(self):
"""Test validation of overly large lists."""
large_list = list(range(1001)) # Exceeds 1000 item limit
with self.assertRaises(ValidationError):
self.client._validate_params(large_list=large_list)
def test_large_dict_validation(self):
"""Test validation of overly large dictionaries."""
large_dict = {f"key_{i}": i for i in range(101)} # Exceeds 100 item limit
with self.assertRaises(ValidationError):
self.client._validate_params(large_dict=large_dict)
def test_valid_parameters_pass(self):
"""Test that valid parameters pass validation."""
# Should not raise any exception
self.client._validate_params(
valid_string="Hello, World!",
valid_list=[1, 2, 3],
valid_dict={"key": "value"},
none_value=None,
)
def test_message_feedback_validation(self):
"""Test validation in message_feedback method."""
with self.assertRaises(ValidationError):
self.client.message_feedback("msg_id", "invalid_rating", "user")
def test_completion_message_validation(self):
"""Test validation in create_completion_message method."""
from dify_client.client import CompletionClient
client = CompletionClient("test_api_key")
with self.assertRaises(ValidationError):
client.create_completion_message(
inputs="not_a_dict", # Should be a dict
response_mode="invalid_mode", # Should be 'blocking' or 'streaming'
user="test_user",
)
def test_chat_message_validation(self):
"""Test validation in create_chat_message method."""
from dify_client.client import ChatClient
client = ChatClient("test_api_key")
with self.assertRaises(ValidationError):
client.create_chat_message(
inputs="not_a_dict", # Should be a dict
query="", # Should not be empty
user="test_user",
response_mode="invalid_mode", # Should be 'blocking' or 'streaming'
)
if __name__ == "__main__":
unittest.main()
+2 -38
View File
@@ -59,7 +59,7 @@ version = "0.1.12"
source = { editable = "." }
dependencies = [
{ name = "aiofiles" },
{ name = "httpx", extra = ["http2"] },
{ name = "httpx" },
]
[package.optional-dependencies]
@@ -71,7 +71,7 @@ dev = [
[package.metadata]
requires-dist = [
{ name = "aiofiles", specifier = ">=23.0.0" },
{ name = "httpx", extras = ["http2"], specifier = ">=0.27.0" },
{ name = "httpx", specifier = ">=0.27.0" },
{ name = "pytest", marker = "extra == 'dev'", specifier = ">=7.0.0" },
{ name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=0.21.0" },
]
@@ -98,28 +98,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/04/4b/29cac41a4d98d144bf5f6d33995617b185d14b22401f75ca86f384e87ff1/h11-0.16.0-py3-none-any.whl", hash = "sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86", size = 37515, upload-time = "2025-04-24T03:35:24.344Z" },
]
[[package]]
name = "h2"
version = "4.3.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "hpack" },
{ name = "hyperframe" },
]
sdist = { url = "https://files.pythonhosted.org/packages/1d/17/afa56379f94ad0fe8defd37d6eb3f89a25404ffc71d4d848893d270325fc/h2-4.3.0.tar.gz", hash = "sha256:6c59efe4323fa18b47a632221a1888bd7fde6249819beda254aeca909f221bf1", size = 2152026, upload-time = "2025-08-23T18:12:19.778Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/69/b2/119f6e6dcbd96f9069ce9a2665e0146588dc9f88f29549711853645e736a/h2-4.3.0-py3-none-any.whl", hash = "sha256:c438f029a25f7945c69e0ccf0fb951dc3f73a5f6412981daee861431b70e2bdd", size = 61779, upload-time = "2025-08-23T18:12:17.779Z" },
]
[[package]]
name = "hpack"
version = "4.1.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/2c/48/71de9ed269fdae9c8057e5a4c0aa7402e8bb16f2c6e90b3aa53327b113f8/hpack-4.1.0.tar.gz", hash = "sha256:ec5eca154f7056aa06f196a557655c5b009b382873ac8d1e66e79e87535f1dca", size = 51276, upload-time = "2025-01-22T21:44:58.347Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/07/c6/80c95b1b2b94682a72cbdbfb85b81ae2daffa4291fbfa1b1464502ede10d/hpack-4.1.0-py3-none-any.whl", hash = "sha256:157ac792668d995c657d93111f46b4535ed114f0c9c8d672271bbec7eae1b496", size = 34357, upload-time = "2025-01-22T21:44:56.92Z" },
]
[[package]]
name = "httpcore"
version = "1.0.9"
@@ -148,20 +126,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/2a/39/e50c7c3a983047577ee07d2a9e53faf5a69493943ec3f6a384bdc792deb2/httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad", size = 73517, upload-time = "2024-12-06T15:37:21.509Z" },
]
[package.optional-dependencies]
http2 = [
{ name = "h2" },
]
[[package]]
name = "hyperframe"
version = "6.1.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/02/e7/94f8232d4a74cc99514c13a9f995811485a6903d48e5d952771ef6322e30/hyperframe-6.1.0.tar.gz", hash = "sha256:f630908a00854a7adeabd6382b43923a4c4cd4b821fcb527e6ab9e15382a3b08", size = 26566, upload-time = "2025-01-22T21:41:49.302Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/48/30/47d0bf6072f7252e6521f3447ccfa40b421b6824517f82854703d0f5a98b/hyperframe-6.1.0-py3-none-any.whl", hash = "sha256:b03380493a519fce58ea5af42e4a42317bf9bd425596f7a0835ffce80f1a42e5", size = 13007, upload-time = "2025-01-22T21:41:47.295Z" },
]
[[package]]
name = "idna"
version = "3.10"
+3 -3
View File
@@ -12,9 +12,6 @@ NEXT_PUBLIC_API_PREFIX=http://localhost:5001/console/api
# console or api domain.
# example: http://udify.app/api
NEXT_PUBLIC_PUBLIC_API_PREFIX=http://localhost:5001/api
# When the frontend and backend run on different subdomains, set NEXT_PUBLIC_COOKIE_DOMAIN=1.
NEXT_PUBLIC_COOKIE_DOMAIN=
# The API PREFIX for MARKETPLACE
NEXT_PUBLIC_MARKETPLACE_API_PREFIX=https://marketplace.dify.ai/api/v1
# The URL for MARKETPLACE
@@ -37,6 +34,9 @@ NEXT_PUBLIC_CSP_WHITELIST=
# Default is not allow to embed into iframe to prevent Clickjacking: https://owasp.org/www-community/attacks/Clickjacking
NEXT_PUBLIC_ALLOW_EMBED=
# Shared cookie domain when console UI and API use different subdomains (e.g. example.com)
NEXT_PUBLIC_COOKIE_DOMAIN=
# Allow rendering unsafe URLs which have "data:" scheme.
NEXT_PUBLIC_ALLOW_UNSAFE_DATA_SCHEME=false
-6
View File
@@ -32,7 +32,6 @@ NEXT_PUBLIC_EDITION=SELF_HOSTED
# different from api or web app domain.
# example: http://cloud.dify.ai/console/api
NEXT_PUBLIC_API_PREFIX=http://localhost:5001/console/api
NEXT_PUBLIC_COOKIE_DOMAIN=
# The URL for Web APP, refers to the Web App base URL of WEB service if web app domain is different from
# console or api domain.
# example: http://udify.app/api
@@ -42,11 +41,6 @@ NEXT_PUBLIC_PUBLIC_API_PREFIX=http://localhost:5001/api
NEXT_PUBLIC_SENTRY_DSN=
```
> [!IMPORTANT]
>
> 1. When the frontend and backend run on different subdomains, set NEXT_PUBLIC_COOKIE_DOMAIN=1. The frontend and backend must be under the same top-level domain in order to share authentication cookies.
> 1. It's necessary to set NEXT_PUBLIC_API_PREFIX and NEXT_PUBLIC_PUBLIC_API_PREFIX to the correct backend API URL.
Finally, run the development server:
```bash
@@ -1,6 +1,6 @@
'use client'
import type { FC } from 'react'
import React, { useCallback, useMemo } from 'react'
import React, { useMemo } from 'react'
import { useTranslation } from 'react-i18next'
import { useContext } from 'use-context-selector'
import AppCard from '@/app/components/app/overview/app-card'
@@ -24,7 +24,6 @@ import { useStore as useAppStore } from '@/app/components/app/store'
import { useAppWorkflow } from '@/service/use-workflow'
import type { BlockEnum } from '@/app/components/workflow/types'
import { isTriggerNode } from '@/app/components/workflow/types'
import { useDocLink } from '@/context/i18n'
export type ICardViewProps = {
appId: string
@@ -34,56 +33,22 @@ export type ICardViewProps = {
const CardView: FC<ICardViewProps> = ({ appId, isInPanel, className }) => {
const { t } = useTranslation()
const docLink = useDocLink()
const { notify } = useContext(ToastContext)
const appDetail = useAppStore(state => state.appDetail)
const setAppDetail = useAppStore(state => state.setAppDetail)
const isWorkflowApp = appDetail?.mode === AppModeEnum.WORKFLOW
const showMCPCard = isInPanel
const showTriggerCard = isInPanel && isWorkflowApp
const { data: currentWorkflow } = useAppWorkflow(isWorkflowApp ? appDetail.id : '')
const hasTriggerNode = useMemo<boolean | null>(() => {
if (!isWorkflowApp)
const showTriggerCard = isInPanel && appDetail?.mode === AppModeEnum.WORKFLOW
const { data: currentWorkflow } = useAppWorkflow(appDetail?.mode === AppModeEnum.WORKFLOW ? appDetail.id : '')
const hasTriggerNode = useMemo(() => {
if (appDetail?.mode !== AppModeEnum.WORKFLOW)
return false
if (!currentWorkflow)
return null
const nodes = currentWorkflow.graph?.nodes || []
const nodes = currentWorkflow?.graph?.nodes || []
return nodes.some((node) => {
const nodeType = node.data?.type as BlockEnum | undefined
return !!nodeType && isTriggerNode(nodeType)
})
}, [isWorkflowApp, currentWorkflow])
const shouldRenderAppCards = !isWorkflowApp || hasTriggerNode === false
const disableAppCards = !shouldRenderAppCards
const triggerDocUrl = docLink('/guides/workflow/node/start')
const buildTriggerModeMessage = useCallback((featureName: string) => (
<div className='flex flex-col gap-1'>
<div className='text-xs text-text-secondary'>
{t('appOverview.overview.disableTooltip.triggerMode', { feature: featureName })}
</div>
<div
className='cursor-pointer text-xs font-medium text-text-accent hover:underline'
onClick={(event) => {
event.stopPropagation()
window.open(triggerDocUrl, '_blank')
}}
>
{t('appOverview.overview.appInfo.enableTooltip.learnMore')}
</div>
</div>
), [t, triggerDocUrl])
const disableWebAppTooltip = disableAppCards
? buildTriggerModeMessage(t('appOverview.overview.appInfo.title'))
: null
const disableApiTooltip = disableAppCards
? buildTriggerModeMessage(t('appOverview.overview.apiInfo.title'))
: null
const disableMcpTooltip = disableAppCards
? buildTriggerModeMessage(t('tools.mcp.server.title'))
: null
}, [appDetail?.mode, currentWorkflow])
const updateAppDetail = async () => {
try {
@@ -155,48 +120,39 @@ const CardView: FC<ICardViewProps> = ({ appId, isInPanel, className }) => {
if (!appDetail)
return <Loading />
const appCards = (
<>
<AppCard
appInfo={appDetail}
cardType="webapp"
isInPanel={isInPanel}
triggerModeDisabled={disableAppCards}
triggerModeMessage={disableWebAppTooltip}
onChangeStatus={onChangeSiteStatus}
onGenerateCode={onGenerateCode}
onSaveSiteConfig={onSaveSiteConfig}
/>
<AppCard
cardType="api"
appInfo={appDetail}
isInPanel={isInPanel}
triggerModeDisabled={disableAppCards}
triggerModeMessage={disableApiTooltip}
onChangeStatus={onChangeApiStatus}
/>
{showMCPCard && (
<MCPServiceCard
appInfo={appDetail}
triggerModeDisabled={disableAppCards}
triggerModeMessage={disableMcpTooltip}
/>
)}
</>
)
const triggerCardNode = showTriggerCard ? (
<TriggerCard
appInfo={appDetail}
onToggleResult={handleCallbackResult}
/>
) : null
return (
<div className={className || 'mb-6 grid w-full grid-cols-1 gap-6 xl:grid-cols-2'}>
{disableAppCards && triggerCardNode}
{appCards}
{!disableAppCards && triggerCardNode}
{
!hasTriggerNode && (
<>
<AppCard
appInfo={appDetail}
cardType="webapp"
isInPanel={isInPanel}
onChangeStatus={onChangeSiteStatus}
onGenerateCode={onGenerateCode}
onSaveSiteConfig={onSaveSiteConfig}
/>
<AppCard
cardType="api"
appInfo={appDetail}
isInPanel={isInPanel}
onChangeStatus={onChangeApiStatus}
/>
{showMCPCard && (
<MCPServiceCard
appInfo={appDetail}
/>
)}
</>
)
}
{showTriggerCard && (
<TriggerCard
appInfo={appDetail}
onToggleResult={handleCallbackResult}
/>
)}
</div>
)
}
@@ -532,7 +532,7 @@ const ProviderConfigModal: FC<Props> = ({
>
<span className='text-[#D92D20]'>{t('common.operation.remove')}</span>
</Button>
<Divider type='vertical' className='mx-3 h-[18px]' />
<Divider className='mx-3 h-[18px]' />
</>
)}
<Button
@@ -1,7 +1,7 @@
'use client'
import { RiArrowLeftLine, RiMailSendFill } from '@remixicon/react'
import { useTranslation } from 'react-i18next'
import { type FormEvent, useCallback, useEffect, useRef, useState } from 'react'
import { useCallback, useState } from 'react'
import { useRouter, useSearchParams } from 'next/navigation'
import { useContext } from 'use-context-selector'
import Countdown from '@/app/components/signin/countdown'
@@ -23,7 +23,6 @@ export default function CheckCode() {
const [code, setVerifyCode] = useState('')
const [loading, setIsLoading] = useState(false)
const { locale } = useContext(I18NContext)
const codeInputRef = useRef<HTMLInputElement>(null)
const redirectUrl = searchParams.get('redirect_url')
const embeddedUserId = useWebAppStore(s => s.embeddedUserId)
@@ -80,15 +79,6 @@ export default function CheckCode() {
}
}
const handleSubmit = (event: FormEvent<HTMLFormElement>) => {
event.preventDefault()
verify()
}
useEffect(() => {
codeInputRef.current?.focus()
}, [])
const resendCode = async () => {
try {
const ret = await sendWebAppEMailLoginCode(email, locale)
@@ -117,18 +107,10 @@ export default function CheckCode() {
</p>
</div>
<form onSubmit={handleSubmit}>
<form action="">
<label htmlFor="code" className='system-md-semibold mb-1 text-text-secondary'>{t('login.checkCode.verificationCode')}</label>
<Input
ref={codeInputRef}
id='code'
value={code}
onChange={e => setVerifyCode(e.target.value)}
maxLength={6}
className='mt-1'
placeholder={t('login.checkCode.verificationCodePlaceholder') || ''}
/>
<Button type='submit' loading={loading} disabled={loading} className='my-3 w-full' variant='primary'>{t('login.checkCode.verify')}</Button>
<Input value={code} onChange={e => setVerifyCode(e.target.value)} maxLength={6} className='mt-1' placeholder={t('login.checkCode.verificationCodePlaceholder') || ''} />
<Button loading={loading} disabled={loading} className='my-3 w-full' variant='primary' onClick={verify}>{t('login.checkCode.verify')}</Button>
<Countdown onResend={resendCode} />
</form>
<div className='py-2'>
+1 -1
View File
@@ -139,7 +139,7 @@ const Annotation: FC<Props> = (props) => {
return (
<div className='flex h-full flex-col'>
<p className='system-sm-regular text-text-tertiary'>{t('appLog.description')}</p>
<div className='relative flex h-full flex-1 flex-col py-4'>
<div className='flex h-full flex-1 flex-col py-4'>
<Filter appId={appDetail.id} queryParams={queryParams} setQueryParams={setQueryParams}>
<div className='flex items-center space-x-2'>
{isChatApp && (
+77 -79
View File
@@ -54,97 +54,95 @@ const List: FC<Props> = ({
}, [isAllSelected, list, selectedIds, onSelectedIdsChange])
return (
<>
<div className='relative mt-2 grow overflow-x-auto'>
<table className={cn('w-full min-w-[440px] border-collapse border-0')}>
<thead className='system-xs-medium-uppercase text-text-tertiary'>
<tr>
<td className='w-12 whitespace-nowrap rounded-l-lg bg-background-section-burn px-2'>
<div className='relative grow overflow-x-auto'>
<table className={cn('mt-2 w-full min-w-[440px] border-collapse border-0')}>
<thead className='system-xs-medium-uppercase text-text-tertiary'>
<tr>
<td className='w-12 whitespace-nowrap rounded-l-lg bg-background-section-burn px-2'>
<Checkbox
className='mr-2'
checked={isAllSelected}
indeterminate={!isAllSelected && isSomeSelected}
onCheck={handleSelectAll}
/>
</td>
<td className='w-5 whitespace-nowrap bg-background-section-burn pl-2 pr-1'>{t('appAnnotation.table.header.question')}</td>
<td className='whitespace-nowrap bg-background-section-burn py-1.5 pl-3'>{t('appAnnotation.table.header.answer')}</td>
<td className='whitespace-nowrap bg-background-section-burn py-1.5 pl-3'>{t('appAnnotation.table.header.createdAt')}</td>
<td className='whitespace-nowrap bg-background-section-burn py-1.5 pl-3'>{t('appAnnotation.table.header.hits')}</td>
<td className='w-[96px] whitespace-nowrap rounded-r-lg bg-background-section-burn py-1.5 pl-3'>{t('appAnnotation.table.header.actions')}</td>
</tr>
</thead>
<tbody className="system-sm-regular text-text-secondary">
{list.map(item => (
<tr
key={item.id}
className='cursor-pointer border-b border-divider-subtle hover:bg-background-default-hover'
onClick={
() => {
onView(item)
}
}
>
<td className='w-12 px-2' onClick={e => e.stopPropagation()}>
<Checkbox
className='mr-2'
checked={isAllSelected}
indeterminate={!isAllSelected && isSomeSelected}
onCheck={handleSelectAll}
checked={selectedIds.includes(item.id)}
onCheck={() => {
if (selectedIds.includes(item.id))
onSelectedIdsChange(selectedIds.filter(id => id !== item.id))
else
onSelectedIdsChange([...selectedIds, item.id])
}}
/>
</td>
<td className='w-5 whitespace-nowrap bg-background-section-burn pl-2 pr-1'>{t('appAnnotation.table.header.question')}</td>
<td className='whitespace-nowrap bg-background-section-burn py-1.5 pl-3'>{t('appAnnotation.table.header.answer')}</td>
<td className='whitespace-nowrap bg-background-section-burn py-1.5 pl-3'>{t('appAnnotation.table.header.createdAt')}</td>
<td className='whitespace-nowrap bg-background-section-burn py-1.5 pl-3'>{t('appAnnotation.table.header.hits')}</td>
<td className='w-[96px] whitespace-nowrap rounded-r-lg bg-background-section-burn py-1.5 pl-3'>{t('appAnnotation.table.header.actions')}</td>
</tr>
</thead>
<tbody className="system-sm-regular text-text-secondary">
{list.map(item => (
<tr
key={item.id}
className='cursor-pointer border-b border-divider-subtle hover:bg-background-default-hover'
onClick={
() => {
onView(item)
}
}
>
<td className='w-12 px-2' onClick={e => e.stopPropagation()}>
<Checkbox
className='mr-2'
checked={selectedIds.includes(item.id)}
onCheck={() => {
if (selectedIds.includes(item.id))
onSelectedIdsChange(selectedIds.filter(id => id !== item.id))
else
onSelectedIdsChange([...selectedIds, item.id])
<td
className='max-w-[250px] overflow-hidden text-ellipsis whitespace-nowrap p-3 pr-2'
title={item.question}
>{item.question}</td>
<td
className='max-w-[250px] overflow-hidden text-ellipsis whitespace-nowrap p-3 pr-2'
title={item.answer}
>{item.answer}</td>
<td className='p-3 pr-2'>{formatTime(item.created_at, t('appLog.dateTimeFormat') as string)}</td>
<td className='p-3 pr-2'>{item.hit_count}</td>
<td className='w-[96px] p-3 pr-2' onClick={e => e.stopPropagation()}>
{/* Actions */}
<div className='flex space-x-1 text-text-tertiary'>
<ActionButton onClick={() => onView(item)}>
<RiEditLine className='h-4 w-4' />
</ActionButton>
<ActionButton
onClick={() => {
setCurrId(item.id)
setShowConfirmDelete(true)
}}
/>
</td>
<td
className='max-w-[250px] overflow-hidden text-ellipsis whitespace-nowrap p-3 pr-2'
title={item.question}
>{item.question}</td>
<td
className='max-w-[250px] overflow-hidden text-ellipsis whitespace-nowrap p-3 pr-2'
title={item.answer}
>{item.answer}</td>
<td className='p-3 pr-2'>{formatTime(item.created_at, t('appLog.dateTimeFormat') as string)}</td>
<td className='p-3 pr-2'>{item.hit_count}</td>
<td className='w-[96px] p-3 pr-2' onClick={e => e.stopPropagation()}>
{/* Actions */}
<div className='flex space-x-1 text-text-tertiary'>
<ActionButton onClick={() => onView(item)}>
<RiEditLine className='h-4 w-4' />
</ActionButton>
<ActionButton
onClick={() => {
setCurrId(item.id)
setShowConfirmDelete(true)
}}
>
<RiDeleteBinLine className='h-4 w-4' />
</ActionButton>
</div>
</td>
</tr>
))}
</tbody>
</table>
<RemoveAnnotationConfirmModal
isShow={showConfirmDelete}
onHide={() => setShowConfirmDelete(false)}
onRemove={() => {
onRemove(currId as string)
setShowConfirmDelete(false)
}}
/>
</div>
>
<RiDeleteBinLine className='h-4 w-4' />
</ActionButton>
</div>
</td>
</tr>
))}
</tbody>
</table>
<RemoveAnnotationConfirmModal
isShow={showConfirmDelete}
onHide={() => setShowConfirmDelete(false)}
onRemove={() => {
onRemove(currId as string)
setShowConfirmDelete(false)
}}
/>
{selectedIds.length > 0 && (
<BatchAction
className='absolute bottom-20 left-0 z-20'
className='absolute bottom-6 left-1/2 z-20 -translate-x-1/2'
selectedIds={selectedIds}
onBatchDelete={onBatchDelete}
onCancel={onCancel}
/>
)}
</>
</div>
)
}
export default React.memo(List)
+22 -52
View File
@@ -49,7 +49,6 @@ import { fetchInstalledAppList } from '@/service/explore'
import { AppModeEnum } from '@/types/app'
import type { PublishWorkflowParams } from '@/types/workflow'
import { basePath } from '@/utils/var'
import UpgradeBtn from '@/app/components/billing/upgrade-btn'
const ACCESS_MODE_MAP: Record<AccessMode, { label: string, icon: React.ElementType }> = {
[AccessMode.ORGANIZATION]: {
@@ -107,7 +106,6 @@ export type AppPublisherProps = {
workflowToolAvailable?: boolean
missingStartNode?: boolean
hasTriggerNode?: boolean // Whether workflow currently contains any trigger nodes (used to hide missing-start CTA when triggers exist).
startNodeLimitExceeded?: boolean
}
const PUBLISH_SHORTCUT = ['ctrl', '⇧', 'P']
@@ -129,7 +127,6 @@ const AppPublisher = ({
workflowToolAvailable = true,
missingStartNode = false,
hasTriggerNode = false,
startNodeLimitExceeded = false,
}: AppPublisherProps) => {
const { t } = useTranslation()
@@ -249,13 +246,6 @@ const AppPublisher = ({
const hasPublishedVersion = !!publishedAt
const workflowToolDisabled = !hasPublishedVersion || !workflowToolAvailable
const workflowToolMessage = workflowToolDisabled ? t('workflow.common.workflowAsToolDisabledHint') : undefined
const showStartNodeLimitHint = Boolean(startNodeLimitExceeded)
const upgradeHighlightStyle = useMemo(() => ({
background: 'linear-gradient(97deg, var(--components-input-border-active-prompt-1, rgba(11, 165, 236, 0.95)) -3.64%, var(--components-input-border-active-prompt-2, rgba(21, 90, 239, 0.95)) 45.14%)',
WebkitBackgroundClip: 'text',
backgroundClip: 'text',
WebkitTextFillColor: 'transparent',
}), [])
return (
<>
@@ -314,49 +304,29 @@ const AppPublisher = ({
/>
)
: (
<>
<Button
variant='primary'
className='mt-3 w-full'
onClick={() => handlePublish()}
disabled={publishDisabled || published}
>
{
published
? t('workflow.common.published')
: (
<div className='flex gap-1'>
<span>{t('workflow.common.publishUpdate')}</span>
<div className='flex gap-0.5'>
{PUBLISH_SHORTCUT.map(key => (
<span key={key} className='system-kbd h-4 w-4 rounded-[4px] bg-components-kbd-bg-white text-text-primary-on-surface'>
{getKeyboardKeyNameBySystem(key)}
</span>
))}
</div>
<Button
variant='primary'
className='mt-3 w-full'
onClick={() => handlePublish()}
disabled={publishDisabled || published}
>
{
published
? t('workflow.common.published')
: (
<div className='flex gap-1'>
<span>{t('workflow.common.publishUpdate')}</span>
<div className='flex gap-0.5'>
{PUBLISH_SHORTCUT.map(key => (
<span key={key} className='system-kbd h-4 w-4 rounded-[4px] bg-components-kbd-bg-white text-text-primary-on-surface'>
{getKeyboardKeyNameBySystem(key)}
</span>
))}
</div>
)
}
</Button>
{showStartNodeLimitHint && (
<div className='mt-3 flex flex-col items-stretch'>
<p
className='text-sm font-semibold leading-5 text-transparent'
style={upgradeHighlightStyle}
>
<span className='block'>{t('workflow.publishLimit.startNodeTitlePrefix')}</span>
<span className='block'>{t('workflow.publishLimit.startNodeTitleSuffix')}</span>
</p>
<p className='mt-1 text-xs leading-4 text-text-secondary'>
{t('workflow.publishLimit.startNodeDesc')}
</p>
<UpgradeBtn
isShort
className='mb-[12px] mt-[9px] h-[32px] w-[93px] self-start'
/>
</div>
)}
</>
</div>
)
}
</Button>
)
}
</div>
+11 -14
View File
@@ -42,7 +42,6 @@ import { getProcessedFilesFromResponse } from '@/app/components/base/file-upload
import cn from '@/utils/classnames'
import { noop } from 'lodash-es'
import PromptLogModal from '../../base/prompt-log-modal'
import { WorkflowContextProvider } from '@/app/components/workflow/context'
type AppStoreState = ReturnType<typeof useAppStore.getState>
type ConversationListItem = ChatConversationGeneralDetail | CompletionConversationGeneralDetail
@@ -780,17 +779,15 @@ function DetailPanel({ detail, onFeedback }: IDetailPanel) {
}
</div>
{showMessageLogModal && (
<WorkflowContextProvider>
<MessageLogModal
width={width}
currentLogItem={currentLogItem}
onCancel={() => {
setCurrentLogItem()
setShowMessageLogModal(false)
}}
defaultTab={currentLogModalActiveTab}
/>
</WorkflowContextProvider>
<MessageLogModal
width={width}
currentLogItem={currentLogItem}
onCancel={() => {
setCurrentLogItem()
setShowMessageLogModal(false)
}}
defaultTab={currentLogModalActiveTab}
/>
)}
{!isChatMode && showPromptLogModal && (
<PromptLogModal
@@ -1030,8 +1027,8 @@ const ConversationList: FC<IConversationList> = ({ logs, appDetail, onRefresh })
return <Loading />
return (
<div className='relative mt-2 grow overflow-x-auto'>
<table className={cn('w-full min-w-[440px] border-collapse border-0')}>
<div className='relative grow overflow-x-auto'>
<table className={cn('mt-2 w-full min-w-[440px] border-collapse border-0')}>
<thead className='system-xs-medium-uppercase text-text-tertiary'>
<tr>
<td className='w-5 whitespace-nowrap rounded-l-lg bg-background-section-burn pl-2 pr-1'></td>
+17 -41
View File
@@ -51,8 +51,6 @@ export type IAppCardProps = {
isInPanel?: boolean
cardType?: 'api' | 'webapp'
customBgColor?: string
triggerModeDisabled?: boolean // true when Trigger Node mode needs UI locked to avoid conflicting actions
triggerModeMessage?: React.ReactNode // contextual copy explaining why the card is disabled in trigger mode
onChangeStatus: (val: boolean) => Promise<void>
onSaveSiteConfig?: (params: ConfigParams) => Promise<void>
onGenerateCode?: () => Promise<void>
@@ -63,8 +61,6 @@ function AppCard({
isInPanel,
cardType = 'webapp',
customBgColor,
triggerModeDisabled = false,
triggerModeMessage = '',
onChangeStatus,
onSaveSiteConfig,
onGenerateCode,
@@ -115,7 +111,7 @@ function AppCard({
const hasStartNode = currentWorkflow?.graph?.nodes?.some(node => node.data.type === BlockEnum.Start)
const missingStartNode = isWorkflowApp && !hasStartNode
const hasInsufficientPermissions = isApp ? !isCurrentWorkspaceEditor : !isCurrentWorkspaceManager
const toggleDisabled = hasInsufficientPermissions || appUnpublished || missingStartNode || triggerModeDisabled
const toggleDisabled = hasInsufficientPermissions || appUnpublished || missingStartNode
const runningStatus = (appUnpublished || missingStartNode) ? false : (isApp ? appInfo.enable_site : appInfo.enable_api)
const isMinimalState = appUnpublished || missingStartNode
const { app_base_url, access_token } = appInfo.site ?? {}
@@ -193,20 +189,7 @@ function AppCard({
className={
`${isInPanel ? 'border-l-[0.5px] border-t' : 'border-[0.5px] shadow-xs'} w-full max-w-full rounded-xl border-effects-highlight ${className ?? ''} ${isMinimalState ? 'h-12' : ''}`}
>
<div className={`${customBgColor ?? 'bg-background-default'} relative rounded-xl ${triggerModeDisabled ? 'opacity-60' : ''}`}>
{triggerModeDisabled && (
triggerModeMessage
? (
<Tooltip
popupContent={triggerModeMessage}
popupClassName="max-w-64 rounded-xl bg-components-panel-bg px-3 py-2 text-xs text-text-secondary shadow-lg"
position="right"
>
<div className='absolute inset-0 z-10 cursor-not-allowed rounded-xl' aria-hidden="true"></div>
</Tooltip>
)
: <div className='absolute inset-0 z-10 cursor-not-allowed rounded-xl' aria-hidden="true"></div>
)}
<div className={`${customBgColor ?? 'bg-background-default'} rounded-xl`}>
<div className={`flex w-full flex-col items-start justify-center gap-3 self-stretch p-3 ${isMinimalState ? 'border-0' : 'border-b-[0.5px] border-divider-subtle'}`}>
<div className='flex w-full items-center gap-3 self-stretch'>
<AppBasic
@@ -231,23 +214,18 @@ function AppCard({
</div>
<Tooltip
popupContent={
toggleDisabled ? (
triggerModeDisabled && triggerModeMessage
? triggerModeMessage
: (appUnpublished || missingStartNode) ? (
<>
<div className="mb-1 text-xs font-normal text-text-secondary">
{t('appOverview.overview.appInfo.enableTooltip.description')}
</div>
<div
className="cursor-pointer text-xs font-normal text-text-accent hover:underline"
onClick={() => window.open(docLink('/guides/workflow/node/user-input'), '_blank')}
>
{t('appOverview.overview.appInfo.enableTooltip.learnMore')}
</div>
</>
)
: ''
toggleDisabled && (appUnpublished || missingStartNode) ? (
<>
<div className="mb-1 text-xs font-normal text-text-secondary">
{t('appOverview.overview.appInfo.enableTooltip.description')}
</div>
<div
className="cursor-pointer text-xs font-normal text-text-accent hover:underline"
onClick={() => window.open(docLink('/guides/workflow/node/user-input'), '_blank')}
>
{t('appOverview.overview.appInfo.enableTooltip.learnMore')}
</div>
</>
) : ''
}
position="right"
@@ -351,11 +329,9 @@ function AppCard({
{!isApp && <SecretKeyButton appId={appInfo.id} />}
{OPERATIONS_MAP[cardType].map((op) => {
const disabled
= triggerModeDisabled
? true
: op.opName === t('appOverview.overview.appInfo.settings.entry')
? false
: !runningStatus
= op.opName === t('appOverview.overview.appInfo.settings.entry')
? false
: !runningStatus
return (
<Button
className="mr-1 min-w-[88px]"
+1 -4
View File
@@ -1,6 +1,6 @@
import React from 'react'
import Link from 'next/link'
import { RiDiscordFill, RiDiscussLine, RiGithubFill } from '@remixicon/react'
import { RiDiscordFill, RiGithubFill } from '@remixicon/react'
import { useTranslation } from 'react-i18next'
type CustomLinkProps = {
@@ -38,9 +38,6 @@ const Footer = () => {
<CustomLink href='https://discord.gg/FngNHpbcY7'>
<RiDiscordFill className='h-5 w-5 text-text-tertiary' />
</CustomLink>
<CustomLink href='https://forum.dify.ai'>
<RiDiscussLine className='h-5 w-5 text-text-tertiary' />
</CustomLink>
</div>
</footer>
)

Some files were not shown because too many files have changed in this diff Show More