Compare commits

..
1251 changed files with 15507 additions and 42727 deletions
@@ -24,7 +24,6 @@ Use this as the component decision guide for Dify web. Existing code is referenc
- Search before adding UI, hooks, helpers, query utilities, or styling patterns. Reuse existing base components, feature components, hooks, utilities, and design styles when they fit.
- Follow Dify's CSS-first Tailwind v4 contract from `packages/dify-ui/README.md` and `packages/dify-ui/AGENTS.md`. Prefer design-system tokens, utilities, and radius mappings over generic Tailwind choices.
- Preserve visible keyboard focus states on the final focusable element. Prefer styled `@langgenius/dify-ui/*` controls when available, because components such as `Button` and form/control primitives carry the standard Dify UI `focus-visible` styling. Do not assume every Dify UI export provides visual focus styles: headless anatomy parts and direct Base UI re-exports such as dialog/popover/tooltip/drawer triggers usually only provide behavior and semantics. When using native `button` / `a`, custom trigger `render` props, clickable rows, icon buttons, menu-like items, or direct trigger parts, verify the rendered focusable element has a visible focus state. If it does not, add the standard Dify UI focus style: `outline-hidden focus-visible:ring-2 focus-visible:ring-state-accent-solid`. Do not hide outlines without an equivalent visible `focus-visible` indicator. Component-specific focus styles should follow an existing styled primitive pattern or a concrete design constraint, not a new ad hoc style.
- Group feature code by workflow, route, or ownership area with route-aligned names: components, hooks, local types, query helpers, atoms, constants, tests, and small utilities should live near the code that changes with them.
- For each feature module, keep a module-local `README.md` as a boundary note. Start with the module name, a brief one-sentence description, then split dependencies into `Internal Modules` and `External Modules` sections; keep both sections and write `None.` when one category is empty. `Internal Modules` lists modules inside the same overall feature using paths from that feature root, such as `shared/domain/runtime-status`; `External Modules` lists project modules outside the feature using paths from the web root without a `web/` prefix, such as `app/components/base/skeleton`. Omit npm packages, workspace package dependencies, and whitelisted plumbing modules. Do not copy caller-relative import paths into the README.
- Module README whitelist: `@/service/client`, `@/next/*`.
@@ -54,7 +53,6 @@ Use this as the component decision guide for Dify web. Existing code is referenc
- Treat `useParams`, route args, and `nuqs` query state as framework-owned state. When atom logic needs those values, hydrate primitive atoms at the route or surface boundary, such as with `useHydrateAtoms(..., { dangerouslyForceHydrate: true })`; keep URL updates in the route/query-state APIs instead of write atoms.
- Within a route-owned feature, choose one source for route identity. If route params are bridged into feature atoms, use that bridge consistently for route-derived queries and actions instead of also threading the same route id through page, tab, and section props.
- For async work tied to atom state, use `atomWithQuery` or `atomWithMutation`; write atoms should update only the inputs that drive those atoms. This applies to pure frontend async work as well as network requests, so do not hand-roll loading/error/in-flight state with `useState` or `useRef` for atom-orchestrated async behavior. For component-owned remote work, use `useQuery` or `useMutation` directly.
- `jotai-tanstack-query` query atoms do not support TanStack Query tracked properties. A component that reads `useAtomValue(queryAtom)` subscribes to the whole query result, even if it only accesses `data`, `isLoading`, or `isError`. Export field-specific derived atoms and have components read the exact fields they render; use `selectAtom(queryAtom, result => result.field)` for query-result fields so unchanged selections do not notify subscribers. Keep direct `useAtomValue(queryAtom)` only when the component or hook genuinely needs the full observer result.
- Row-local async state belongs to the row owner unless it participates in a shared Jotai workflow or needs atom-scoped reset semantics.
- Leave query and mutation atoms unscoped so they keep shared QueryClient cache and invalidation behavior. Scope resettable primitives and explicit hydration tuples; scope a derived atom only when every dependency should be private to that surface.
- For scoped primitives that are always hydrated by `ScopeProvider`, prefer `atomWithLazy<T>(() => { throw new Error(...) })` when consumers should see a non-null type.
-9
View File
@@ -53,10 +53,6 @@ jobs:
filters: |
api:
- 'api/**'
- 'scripts/check_no_new_getattr.py'
- 'scripts/ast_grep_rules/no_new_getattr.yml'
- '.github/workflows/style.yml'
- '.github/workflows/main-ci.yml'
- '.github/workflows/api-tests.yml'
- 'docker/.env.example'
- 'docker/envs/middleware.env.example'
@@ -99,7 +95,6 @@ jobs:
- '.nvmrc'
- 'docker/docker-compose.middleware.yaml'
- 'docker/envs/middleware.env.example'
- '.github/workflows/main-ci.yml'
- '.github/workflows/web-e2e.yml'
- '.github/actions/setup-web/**'
vdb:
@@ -327,8 +322,6 @@ jobs:
- check-changes
if: needs.pre_job.outputs.should_skip != 'true' && needs.check-changes.outputs.e2e-changed == 'true'
uses: ./.github/workflows/web-e2e.yml
with:
run-external-runtime: true
secrets: inherit
web-e2e-skip:
@@ -387,8 +380,6 @@ jobs:
needs: pre_job
if: needs.pre_job.outputs.should_skip != 'true'
uses: ./.github/workflows/style.yml
with:
base-rev: ${{ github.event.pull_request.base.sha || github.event.merge_group.base_sha }}
vdb-tests-run:
name: Run VDB Tests
+9 -6
View File
@@ -2,10 +2,6 @@ name: Style check
on:
workflow_call:
inputs:
base-rev:
required: true
type: string
concurrency:
group: style-${{ github.head_ref || github.run_id }}
@@ -37,7 +33,6 @@ jobs:
scripts/check_no_new_getattr.py
scripts/ast_grep_rules/no_new_getattr.yml
.github/workflows/style.yml
.github/workflows/main-ci.yml
- name: Setup UV and Python
if: steps.changed-files.outputs.any_changed == 'true'
@@ -59,9 +54,17 @@ jobs:
if: steps.changed-files.outputs.any_changed == 'true'
run: uv run --project api --dev python api/dev/lint_response_contracts.py --fail-on-mismatch
- name: Fetch merge target ref for getattr guard
if: steps.changed-files.outputs.any_changed == 'true'
run: git fetch --no-tags --depth=1 origin +refs/heads/main:refs/remotes/origin/main
- name: Bind merge target branch for getattr guard
if: steps.changed-files.outputs.any_changed == 'true'
run: git show-ref --verify --quiet refs/heads/main || git branch main origin/main
- name: Run No New Getattr Guard
if: steps.changed-files.outputs.any_changed == 'true'
run: uv run --project api python scripts/check_no_new_getattr.py --base-rev "${{ inputs.base-rev }}"
run: uv run --project api python scripts/check_no_new_getattr.py --mode ci --merge-target main
- name: Run Type Checks
if: steps.changed-files.outputs.any_changed == 'true'
+1 -10
View File
@@ -96,15 +96,6 @@ jobs:
vp run e2e:external:prepare
vp run e2e:external
- name: Print E2E log tails
if: ${{ failure() && inputs.run-external-runtime }}
run: |
while IFS= read -r log_file; do
echo "::group::${log_file}"
tail -n 200 "${log_file}"
echo "::endgroup::"
done < <(find e2e/.logs -type f | sort)
- name: Upload Cucumber report
if: ${{ !cancelled() }}
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
@@ -120,5 +111,5 @@ jobs:
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
with:
name: e2e-logs
path: e2e/.logs/**
path: e2e/.logs
retention-days: 7
@@ -0,0 +1,30 @@
# HITL timeout semantics implementation report
## What changed
- Updated `api/core/workflow/nodes/human_input/callback.py` so `DifyHITLCallback` now preserves Dify's timeout split at the boundary:
- `HumanInputFormStatus.TIMEOUT` returns the graphon timeout branch via `Expired(selected_handle="__timeout__", ...)`.
- `HumanInputFormStatus.EXPIRED` is treated as an invalid resume state and raises `AssertionError`.
- `HumanInputFormStatus.WAITING` with a past global deadline is treated as an invalid resume state and raises `AssertionError`.
- `HumanInputFormStatus.WAITING` with only the node-level deadline expired still returns the timeout branch.
- Added `created_at` to `HumanInputFormEntity` and `_HumanInputFormEntityImpl` so the callback can compute the global deadline using Dify's shared `HUMAN_INPUT_GLOBAL_TIMEOUT_SECONDS` invariant.
- Kept the submitted and pause flows unchanged.
- Added focused unit coverage in `api/tests/unit_tests/core/workflow/test_human_input_callback.py` for:
- node timeout branch
- global expiration rejection
- waiting-form past node deadline timeout
- waiting-form past global deadline rejection
## Verification
- `uv run --project api pytest -o addopts='' api/tests/unit_tests/core/workflow/test_human_input_callback.py api/tests/unit_tests/core/workflow/nodes/human_input/test_human_input_form_filled_event.py -q`
- `git diff --check`
## Result
- The focused test set is expected to pass with the new `created_at` boundary in place.
- No unrelated files were modified.
## Concerns
- The callback now fails fast on invalid resume states by design. That is intentional, but any caller that previously relied on `EXPIRED` being mapped to the timeout branch will now see an assertion failure instead.
+2
View File
@@ -125,6 +125,8 @@ All of Dify's offerings come with corresponding APIs, so you could effortlessly
- **Dify for enterprise / organizations<br/>**
We provide additional enterprise-centric features. [Send us an email](mailto:business@dify.ai?subject=%5BGitHub%5DBusiness%20License%20Inquiry) to discuss your enterprise needs. <br/>
> For startups and small businesses using AWS, check out [Dify Premium on AWS Marketplace](https://aws.amazon.com/marketplace/pp/prodview-t22mebxzwjhu6) and deploy it to your own AWS VPC with one click. It's an affordable AMI offering with the option to create apps with custom logo and branding.
## Staying ahead
Star Dify on GitHub and be instantly notified of new releases.
-3
View File
@@ -663,9 +663,6 @@ PLUGIN_MODEL_SCHEMA_CACHE_TTL=3600
PLUGIN_MODEL_PROVIDERS_CACHE_TTL=86400
INNER_API_KEY_FOR_PLUGIN=QaHbTe77CtuXmsfyhR7+vRjI/+XbV1AaFy691iy+kGDv2Jvy0/eAh8Y1
# Dify Agent backend
AGENT_BACKEND_BASE_URL=http://localhost:5050
# Marketplace configuration
MARKETPLACE_ENABLED=true
MARKETPLACE_API_URL=https://marketplace.dify.ai
-10
View File
@@ -18,7 +18,6 @@ from clients.agent_backend.errors import (
AgentBackendValidationError,
)
from clients.agent_backend.event_adapter import (
AgentBackendAgentMessageDeltaInternalEvent,
AgentBackendDeferredToolCallInternalEvent,
AgentBackendInternalEvent,
AgentBackendInternalEventType,
@@ -47,11 +46,6 @@ from clients.agent_backend.request_builder import (
AgentBackendWorkflowNodeRunInput,
redact_for_agent_backend_log,
)
from clients.agent_backend.session_cleanup import (
AgentBackendSessionCleanupPayload,
AgentBackendSessionCleanupResult,
cleanup_agent_backend_session,
)
__all__ = [
"AGENT_SOUL_PROMPT_LAYER_ID",
@@ -63,7 +57,6 @@ __all__ = [
"WORKFLOW_NODE_JOB_PROMPT_LAYER_ID",
"WORKFLOW_USER_PROMPT_LAYER_ID",
"AgentBackendAgentAppRunInput",
"AgentBackendAgentMessageDeltaInternalEvent",
"AgentBackendDeferredToolCallInternalEvent",
"AgentBackendError",
"AgentBackendHTTPError",
@@ -80,8 +73,6 @@ __all__ = [
"AgentBackendRunRequestBuilder",
"AgentBackendRunStartedInternalEvent",
"AgentBackendRunSucceededInternalEvent",
"AgentBackendSessionCleanupPayload",
"AgentBackendSessionCleanupResult",
"AgentBackendStreamError",
"AgentBackendStreamInternalEvent",
"AgentBackendTransportError",
@@ -91,7 +82,6 @@ __all__ = [
"FakeAgentBackendRunClient",
"FakeAgentBackendScenario",
"RuntimeLayerSpec",
"cleanup_agent_backend_session",
"create_agent_backend_run_client",
"extract_runtime_layer_specs",
"redact_for_agent_backend_log",
@@ -5,9 +5,6 @@ The adapter does not define a new cross-service event contract. It consumes
workflow Agent Node maps to Graphon/AppQueue events. Deferred external tool calls
remain Dify Agent ``run_succeeded`` payloads on the wire; API code turns them
into an internal event so workflow pause/session handling stays local to API.
Agent-message deltas are exposed as annotations on ``PydanticAIStreamRunEvent``
so API code does not have to parse Pydantic AI stream-event internals to
preserve streaming. The terminal answer remains the ``run_succeeded`` output.
"""
from __future__ import annotations
@@ -35,7 +32,6 @@ class AgentBackendInternalEventType(StrEnum):
RUN_STARTED = "run_started"
STREAM_EVENT = "stream_event"
AGENT_MESSAGE_DELTA = "agent_message_delta"
DEFERRED_TOOL_CALL = "deferred_tool_call"
RUN_SUCCEEDED = "run_succeeded"
RUN_FAILED = "run_failed"
@@ -65,13 +61,6 @@ class AgentBackendStreamInternalEvent(AgentBackendInternalEventBase):
data: JsonValue
class AgentBackendAgentMessageDeltaInternalEvent(AgentBackendInternalEventBase):
"""API-internal agent-message delta emitted independently from raw stream events."""
type: Literal[AgentBackendInternalEventType.AGENT_MESSAGE_DELTA] = AgentBackendInternalEventType.AGENT_MESSAGE_DELTA
delta: str
class AgentBackendRunSucceededInternalEvent(AgentBackendInternalEventBase):
"""API-internal terminal success event carrying final output and session state."""
@@ -110,7 +99,6 @@ class AgentBackendRunCancelledInternalEvent(AgentBackendInternalEventBase):
type AgentBackendInternalEvent = Annotated[
AgentBackendRunStartedInternalEvent
| AgentBackendStreamInternalEvent
| AgentBackendAgentMessageDeltaInternalEvent
| AgentBackendDeferredToolCallInternalEvent
| AgentBackendRunSucceededInternalEvent
| AgentBackendRunFailedInternalEvent
@@ -133,14 +121,6 @@ class AgentBackendRunEventAdapter:
)
]
case PydanticAIStreamRunEvent():
if event.agent_message_delta:
return [
AgentBackendAgentMessageDeltaInternalEvent(
run_id=event.run_id,
source_event_id=event.id,
delta=event.agent_message_delta,
)
]
data = cast(JsonValue, _EVENT_DATA_ADAPTER.dump_python(event.data, mode="json"))
event_kind = data.get("event_kind") if isinstance(data, dict) else None
return [
+11 -70
View File
@@ -11,9 +11,8 @@ composition-driven.
from __future__ import annotations
import re
from collections.abc import Mapping
from typing import ClassVar, Literal
from typing import ClassVar
from agenton.compositor import CompositorSessionSnapshot
from agenton.compositor.schemas import LayerSessionSnapshot
@@ -47,6 +46,7 @@ from dify_agent.protocol import (
LayerExitSignals,
RunComposition,
RunLayerSpec,
RunPurpose,
RuntimeLayerSpec,
)
from pydantic import BaseModel, ConfigDict, Field, JsonValue, field_validator
@@ -63,7 +63,6 @@ DIFY_CORE_TOOLS_LAYER_ID = "core_tools"
DIFY_KNOWLEDGE_BASE_LAYER_ID = "knowledge"
DIFY_ASK_HUMAN_LAYER_ID = "ask_human"
DIFY_SHELL_LAYER_ID = "shell"
type AgentConfigVersionKind = Literal["snapshot", "draft", "build_draft"]
def _filter_snapshot_to_specs(
@@ -105,59 +104,6 @@ def _shell_config_with_drive_ref(
return config.model_copy(update={"agent_stub_drive_ref": drive_config.drive_ref})
def _markdown_backtick_fence(text: str) -> str:
"""Choose a fence that will not terminate inside the prompt body."""
longest_backtick_run = max((len(match.group(0)) for match in re.finditer(r"`+", text)), default=0)
return "`" * max(3, longest_backtick_run + 1)
_BUILD_DRAFT_AGENT_SOUL_PROMPT = """You are running in build mode.
Objective:
- Improve this agent's working environment, configuration, tools, files, notes,
and context so it can handle the intended task well.
Guidance:
- Treat the intended task as context for setup work, validation, and configuration decisions.
- Perform concrete investigative or setup steps when they help improve or verify the agent configuration.
- Use the installed `dify-agent` CLI when you need to inspect or persist Agent configuration."""
def _wrap_build_draft_agent_soul_prompt(prompt: str | None) -> str:
"""Reframe build-draft Agent Soul prompts as preparation work for a future run."""
prompt_body = (prompt or "").strip()
if not prompt_body:
return _BUILD_DRAFT_AGENT_SOUL_PROMPT + "\n\nIntended task for later normal runs:\nNo task prompt was provided."
fence = _markdown_backtick_fence(prompt_body)
return (
_BUILD_DRAFT_AGENT_SOUL_PROMPT
+ f"\n\nIntended task for later normal runs:\n{fence}text\n{prompt_body}\n{fence}"
)
def _agent_soul_prompt_for_layer(
prompt: str | None,
*,
config_version_kind: AgentConfigVersionKind,
) -> str | None:
"""Preserve normal snapshot/draft prompts and only wrap build-draft prompts.
The API-side layer adapter is the product boundary where Agent Soul text
becomes the model-facing system-prompt layer. ``snapshot`` and normal
``draft`` runs pass through the original effective prompt unchanged, while
``build_draft`` always emits a setup prompt. When an original prompt is
present, it is reframed as future-run context and embedded in a fenced
block; when it is blank, the setup instruction is still kept.
"""
if config_version_kind != "build_draft":
if prompt is None:
return None
if not prompt.strip():
return None
return prompt
return _wrap_build_draft_agent_soul_prompt(prompt)
class AgentBackendModelConfig(BaseModel):
"""API-side model/plugin selection before it is converted to Dify Agent layers."""
@@ -217,7 +163,7 @@ class AgentBackendWorkflowNodeRunInput(BaseModel):
workflow_node_job_prompt: str
user_prompt: str
agent_soul_prompt: str | None = None
agent_config_version_kind: AgentConfigVersionKind = "snapshot"
purpose: RunPurpose = "workflow_node"
idempotency_key: str | None = None
output: AgentBackendOutputConfig | None = None
tools: DifyPluginToolsLayerConfig | None = None
@@ -266,7 +212,7 @@ class AgentBackendAgentAppRunInput(BaseModel):
execution_context: DifyExecutionContextLayerConfig
user_prompt: str
agent_soul_prompt: str | None = None
agent_config_version_kind: AgentConfigVersionKind = "snapshot"
purpose: RunPurpose = "agent_app"
idempotency_key: str | None = None
output: AgentBackendOutputConfig | None = None
tools: DifyPluginToolsLayerConfig | None = None
@@ -315,17 +261,13 @@ class AgentBackendRunRequestBuilder:
prompt.
"""
layers: list[RunLayerSpec] = []
agent_soul_prompt = _agent_soul_prompt_for_layer(
run_input.agent_soul_prompt,
config_version_kind=run_input.agent_config_version_kind,
)
if agent_soul_prompt:
if run_input.agent_soul_prompt:
layers.append(
RunLayerSpec(
name=AGENT_SOUL_PROMPT_LAYER_ID,
type=PLAIN_PROMPT_LAYER_TYPE_ID,
metadata={**run_input.metadata, "origin": "agent_soul"},
config=PromptLayerConfig(prefix=agent_soul_prompt),
config=PromptLayerConfig(prefix=run_input.agent_soul_prompt),
)
)
@@ -477,6 +419,7 @@ class AgentBackendRunRequestBuilder:
return CreateRunRequest(
composition=RunComposition(layers=layers),
purpose=run_input.purpose,
idempotency_key=run_input.idempotency_key,
metadata=run_input.metadata,
session_snapshot=run_input.session_snapshot,
@@ -524,6 +467,7 @@ class AgentBackendRunRequestBuilder:
filtered_snapshot = _filter_snapshot_to_specs(session_snapshot, runtime_layer_specs)
return CreateRunRequest(
composition=RunComposition(layers=layers),
purpose="workflow_node",
idempotency_key=idempotency_key,
metadata=request_metadata,
session_snapshot=filtered_snapshot,
@@ -539,17 +483,13 @@ class AgentBackendRunRequestBuilder:
ask_human / structured output.
"""
layers: list[RunLayerSpec] = []
agent_soul_prompt = _agent_soul_prompt_for_layer(
run_input.agent_soul_prompt,
config_version_kind=run_input.agent_config_version_kind,
)
if agent_soul_prompt:
if run_input.agent_soul_prompt:
layers.append(
RunLayerSpec(
name=AGENT_SOUL_PROMPT_LAYER_ID,
type=PLAIN_PROMPT_LAYER_TYPE_ID,
metadata={**run_input.metadata, "origin": "agent_soul"},
config=PromptLayerConfig(prefix=agent_soul_prompt),
config=PromptLayerConfig(prefix=run_input.agent_soul_prompt),
)
)
@@ -709,6 +649,7 @@ class AgentBackendRunRequestBuilder:
return CreateRunRequest(
composition=RunComposition(layers=layers),
purpose=run_input.purpose,
idempotency_key=run_input.idempotency_key,
metadata=run_input.metadata,
session_snapshot=run_input.session_snapshot,
@@ -1,100 +0,0 @@
"""Shared API-side helper for Agent backend lifecycle-only session cleanup.
Product code owns local row retirement and background-task dispatch. This module
only adapts persisted cleanup inputs into the public ``dify-agent`` run
protocol, performs the synchronous ``create_run + wait_run`` loop used by Celery
workers, and reports whether the backend cleanup succeeded, was skipped, or
failed.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import ClassVar, Literal
from agenton.compositor import CompositorSessionSnapshot
from dify_agent.protocol import RuntimeLayerSpec
from pydantic import BaseModel, ConfigDict, Field, JsonValue
from clients.agent_backend.client import AgentBackendRunClient
from clients.agent_backend.errors import AgentBackendError
from clients.agent_backend.request_builder import AgentBackendRunRequestBuilder
class AgentBackendSessionCleanupPayload(BaseModel):
"""Serialized cleanup inputs preserved across API and Celery boundaries."""
session_snapshot: CompositorSessionSnapshot | None = None
runtime_layer_specs: list[RuntimeLayerSpec] = Field(default_factory=list)
idempotency_key: str | None = None
metadata: dict[str, JsonValue] = Field(default_factory=dict)
timeout_seconds: float = 30.0
model_config: ClassVar[ConfigDict] = ConfigDict(extra="forbid")
@dataclass(frozen=True, slots=True)
class AgentBackendSessionCleanupResult:
"""Terminal outcome of one backend cleanup attempt."""
status: Literal["succeeded", "skipped", "failed"]
reason: str | None = None
cleanup_run_id: str | None = None
@classmethod
def succeeded(cls, cleanup_run_id: str) -> AgentBackendSessionCleanupResult:
return cls(status="succeeded", cleanup_run_id=cleanup_run_id)
@classmethod
def skipped(cls, reason: str) -> AgentBackendSessionCleanupResult:
return cls(status="skipped", reason=reason)
@classmethod
def failed(cls, reason: str, cleanup_run_id: str | None = None) -> AgentBackendSessionCleanupResult:
return cls(status="failed", reason=reason, cleanup_run_id=cleanup_run_id)
def cleanup_agent_backend_session(
*,
payload: AgentBackendSessionCleanupPayload,
client: AgentBackendRunClient | None,
request_builder: AgentBackendRunRequestBuilder | None = None,
) -> AgentBackendSessionCleanupResult:
"""Run lifecycle-only cleanup against the Agent backend and report status."""
if client is None:
return AgentBackendSessionCleanupResult.skipped("no_agent_backend_client")
if payload.session_snapshot is None:
return AgentBackendSessionCleanupResult.skipped("missing_session_snapshot")
if not payload.runtime_layer_specs:
return AgentBackendSessionCleanupResult.skipped("missing_runtime_layer_specs")
builder = request_builder or AgentBackendRunRequestBuilder()
request = builder.build_cleanup_request(
session_snapshot=payload.session_snapshot,
runtime_layer_specs=payload.runtime_layer_specs,
idempotency_key=payload.idempotency_key,
metadata=payload.metadata,
)
try:
response = client.create_run(request)
except AgentBackendError as exc:
return AgentBackendSessionCleanupResult.failed(str(exc))
try:
status_response = client.wait_run(response.run_id, timeout_seconds=payload.timeout_seconds)
except AgentBackendError as exc:
return AgentBackendSessionCleanupResult.failed(str(exc), cleanup_run_id=response.run_id)
if status_response.status != "succeeded":
reason = status_response.error or f"cleanup run ended with status {status_response.status}"
return AgentBackendSessionCleanupResult.failed(reason, cleanup_run_id=response.run_id)
return AgentBackendSessionCleanupResult.succeeded(response.run_id)
__all__ = [
"AgentBackendSessionCleanupPayload",
"AgentBackendSessionCleanupResult",
"cleanup_agent_backend_session",
]
+2 -2
View File
@@ -25,7 +25,7 @@ def reset_password(email, new_password, password_confirm):
return
normalized_email = email.strip().lower()
account = AccountService.get_account_by_email_with_case_fallback(email.strip(), session=db.session())
account = AccountService.get_account_by_email_with_case_fallback(db.session(), email.strip())
if not account:
click.echo(click.style(f"Account not found for email: {email}", fg="red"))
@@ -67,7 +67,7 @@ def reset_email(email, new_email, email_confirm):
return
normalized_new_email = new_email.strip().lower()
account = AccountService.get_account_by_email_with_case_fallback(email.strip(), session=db.session())
account = AccountService.get_account_by_email_with_case_fallback(db.session(), email.strip())
if not account:
click.echo(click.style(f"Account not found for email: {email}", fg="red"))
+36 -122
View File
@@ -1,12 +1,10 @@
import datetime
import logging
import time
from collections.abc import Callable
from typing import TypedDict
import click
import sqlalchemy as sa
from sqlalchemy.orm import Session, sessionmaker
from extensions.ext_database import db
from libs.datetime_utils import naive_utc_now
@@ -14,7 +12,6 @@ from services.clear_free_plan_tenant_expired_logs import ClearFreePlanTenantExpi
from services.retention.conversation.messages_clean_policy import create_message_clean_policy
from services.retention.conversation.messages_clean_service import MessagesCleanService
from services.retention.workflow_run.clear_free_plan_expired_workflow_run_logs import WorkflowRunCleanup
from services.retention.workflow_run.db_retry import run_with_db_retry
from services.retention.workflow_run.tenant_prefix import tenant_prefix_condition
from tasks.remove_app_and_related_data_task import delete_draft_variables_batch
@@ -38,12 +35,6 @@ class WorkflowRunArchiveTenantPlan(TypedDict):
unpaid_tenant_ids: list[str]
class WorkflowRunArchivePrefixStats(TypedDict):
tenant_ids: list[str]
workflow_runs: int
workflow_node_executions: int
def _normalize_utc_datetime(value: datetime.datetime) -> datetime.datetime:
if value.tzinfo is None:
return value.replace(tzinfo=datetime.UTC)
@@ -66,7 +57,6 @@ def _parse_tenant_prefixes(prefixes: str | None) -> list[str]:
def _get_archive_candidate_tenant_ids_by_prefix(
session: Session,
prefix: str,
*,
start_from: datetime.datetime | None,
@@ -85,7 +75,7 @@ def _get_archive_candidate_tenant_ids_by_prefix(
if start_from is not None:
conditions.append(WorkflowRun.created_at >= start_from)
tenant_ids = session.scalars(
tenant_ids = db.session.scalars(
sa.select(WorkflowRun.tenant_id).where(*conditions).distinct().order_by(WorkflowRun.tenant_id)
).all()
return list(tenant_ids)
@@ -112,80 +102,8 @@ def _filter_paid_workflow_archive_tenant_ids(tenant_ids: list[str]) -> tuple[lis
return paid_tenant_ids, unpaid_tenant_ids
def _run_archive_command_db_retry[T](operation_name: str, operation: Callable[[], T]) -> T:
return run_with_db_retry(operation_name, operation, logger=logger)
def _get_archive_candidate_tenant_ids_with_retry(
session_maker: sessionmaker[Session],
prefix: str,
*,
start_from: datetime.datetime | None,
end_before: datetime.datetime,
) -> list[str]:
def fetch_tenant_ids() -> list[str]:
with session_maker() as session:
return _get_archive_candidate_tenant_ids_by_prefix(
session,
prefix,
start_from=start_from,
end_before=end_before,
)
return _run_archive_command_db_retry(f"workflow archive tenant resolve for prefix {prefix}", fetch_tenant_ids)
def _get_archive_plan_prefix_stats(
session_maker: sessionmaker[Session],
prefix: str,
*,
start_from: datetime.datetime | None,
end_before: datetime.datetime,
) -> WorkflowRunArchivePrefixStats:
from graphon.enums import WorkflowExecutionStatus
from models.workflow import WorkflowNodeExecutionModel, WorkflowRun
from services.retention.workflow_run.archive_paid_plan_workflow_run import WorkflowRunArchiver
def fetch_prefix_stats() -> WorkflowRunArchivePrefixStats:
with session_maker() as session:
tenant_ids = _get_archive_candidate_tenant_ids_by_prefix(
session,
prefix,
start_from=start_from,
end_before=end_before,
)
run_conditions = [
WorkflowRun.created_at < end_before,
WorkflowRun.status.in_(WorkflowExecutionStatus.ended_values()),
WorkflowRun.type.in_(WorkflowRunArchiver.ARCHIVED_TYPE),
tenant_prefix_condition(WorkflowRun.tenant_id, prefix),
]
if start_from is not None:
run_conditions.append(WorkflowRun.created_at >= start_from)
workflow_runs = (
session.scalar(sa.select(sa.func.count()).select_from(WorkflowRun).where(*run_conditions)) or 0
)
candidate_runs = sa.select(WorkflowRun.id).where(*run_conditions).subquery()
workflow_node_executions = (
session.scalar(
sa.select(sa.func.count())
.select_from(WorkflowNodeExecutionModel)
.join(candidate_runs, WorkflowNodeExecutionModel.workflow_run_id == candidate_runs.c.id)
)
or 0
)
return WorkflowRunArchivePrefixStats(
tenant_ids=tenant_ids,
workflow_runs=workflow_runs,
workflow_node_executions=workflow_node_executions,
)
return _run_archive_command_db_retry(f"workflow archive plan for prefix {prefix}", fetch_prefix_stats)
def _resolve_archive_tenant_ids_from_plan(
*,
session_maker: sessionmaker[Session],
tenant_ids: str | None,
tenant_prefixes: list[str],
start_from: datetime.datetime | None,
@@ -204,8 +122,7 @@ def _resolve_archive_tenant_ids_from_plan(
requested_tenant_ids = []
for prefix in tenant_prefixes:
requested_tenant_ids.extend(
_get_archive_candidate_tenant_ids_with_retry(
session_maker,
_get_archive_candidate_tenant_ids_by_prefix(
prefix,
start_from=start_from,
end_before=end_before,
@@ -226,21 +143,6 @@ def _resolve_archive_tenant_ids_from_plan(
)
def _safe_remove_scoped_session(context: str) -> None:
try:
db.session.remove()
except Exception:
logger.warning("Ignoring DB scoped-session cleanup error after %s", context, exc_info=True)
try:
db.session.registry.clear()
except Exception:
logger.warning("Ignoring DB scoped-session registry cleanup error after %s", context, exc_info=True)
try:
db.engine.dispose()
except Exception:
logger.warning("Ignoring DB engine dispose error after %s", context, exc_info=True)
def _resolve_archive_time_range(
*,
before_days: int,
@@ -447,6 +349,10 @@ def archive_workflow_runs_plan(
supported workflow types, and the requested created_at window. V2 bundle archive
does not maintain per-run archive logs, so this plan reports source-table volume.
"""
from graphon.enums import WorkflowExecutionStatus
from models.workflow import WorkflowNodeExecutionModel, WorkflowRun
from services.retention.workflow_run.archive_paid_plan_workflow_run import WorkflowRunArchiver
before_days, start_from, end_before = _resolve_archive_time_range(
before_days=before_days,
from_days_ago=from_days_ago,
@@ -458,25 +364,37 @@ def archive_workflow_runs_plan(
if include_archived:
click.echo(click.style("--include-archived is a no-op for V2 bundle archive plans.", fg="yellow"))
session_maker = sessionmaker(bind=db.engine, expire_on_commit=False)
rows: list[WorkflowRunArchivePlanRow] = []
for prefix in _HEX_PREFIXES:
try:
prefix_stats = _get_archive_plan_prefix_stats(
session_maker,
prefix,
start_from=start_from,
end_before=plan_end_before,
)
except Exception as exc:
logger.exception("Failed to build workflow archive plan for prefix %s", prefix)
raise click.ClickException(f"Failed to build workflow archive plan for prefix {prefix}.") from exc
tenant_ids = prefix_stats["tenant_ids"]
workflow_runs = prefix_stats["workflow_runs"]
workflow_node_executions = prefix_stats["workflow_node_executions"]
tenant_ids = _get_archive_candidate_tenant_ids_by_prefix(
prefix,
start_from=start_from,
end_before=plan_end_before,
)
total_tenants = len(tenant_ids)
paid_tenant_ids, unpaid_tenant_ids = _filter_paid_workflow_archive_tenant_ids(tenant_ids)
run_conditions = [
WorkflowRun.created_at < plan_end_before,
WorkflowRun.status.in_(WorkflowExecutionStatus.ended_values()),
WorkflowRun.type.in_(WorkflowRunArchiver.ARCHIVED_TYPE),
tenant_prefix_condition(WorkflowRun.tenant_id, prefix),
]
if start_from is not None:
run_conditions.append(WorkflowRun.created_at >= start_from)
workflow_runs = (
db.session.scalar(sa.select(sa.func.count()).select_from(WorkflowRun).where(*run_conditions)) or 0
)
candidate_runs = sa.select(WorkflowRun.id).where(*run_conditions).subquery()
workflow_node_executions = (
db.session.scalar(
sa.select(sa.func.count())
.select_from(WorkflowNodeExecutionModel)
.join(candidate_runs, WorkflowNodeExecutionModel.workflow_run_id == candidate_runs.c.id)
)
or 0
)
rows.append(
WorkflowRunArchivePlanRow(
tenant_prefix=prefix,
@@ -656,18 +574,17 @@ def archive_workflow_runs(
)
)
session_maker = sessionmaker(bind=db.engine, expire_on_commit=False)
try:
tenant_plan = _resolve_archive_tenant_ids_from_plan(
session_maker=session_maker,
tenant_ids=tenant_ids,
tenant_prefixes=parsed_tenant_prefixes,
start_from=start_from,
end_before=plan_end_before,
)
except Exception as exc:
except Exception:
logger.exception("Failed to resolve workflow archive tenant plan")
raise click.ClickException("Failed to resolve workflow archive tenant plan.") from exc
click.echo(click.style("Failed to resolve workflow archive tenant plan.", fg="red"))
return
planned_tenant_ids = tenant_plan["archive_tenant_ids"]
planned_paid_tenant_ids = tenant_plan["paid_tenant_ids"] if planned_tenant_ids is not None else None
@@ -699,10 +616,7 @@ def archive_workflow_runs(
dry_run=dry_run,
delete_after_archive=delete_after_archive,
)
try:
summary = archiver.run()
finally:
_safe_remove_scoped_session("archive workflow run command")
summary = archiver.run()
click.echo(
click.style(
f"Summary: processed={summary.total_runs_processed}, archived={summary.runs_archived}, "
+4 -3
View File
@@ -25,10 +25,11 @@ class AgentBackendConfig(BaseSettings):
AGENT_SHELL_ENABLED: bool = Field(
description=(
"Inject the dify.shell layer (sandboxed bash workspace) into Agent runs. "
"Requires the agent backend to be wired with a shellctl entrypoint before "
"shell-using Agent runs are executed."
"Requires the agent backend to be wired with a shellctl entrypoint; keep it "
"off until shellctl is deployed, otherwise every agent run that includes the "
"shell layer will fail."
),
default=True,
default=False,
)
AGENT_APP_TEXT_DELTA_DEBOUNCE_SECONDS: NonNegativeFloat = Field(
+1 -4
View File
@@ -363,10 +363,7 @@ class FileAccessConfig(BaseSettings):
INTERNAL_FILES_URL: str = Field(
description="Internal base URL for file access within Docker network,"
" used for plugin daemon and internal service communication."
" Explicit INTERNAL_FILES_URL takes precedence; otherwise SERVER_CONSOLE_API_URL is used,"
" then FILES_URL.",
validation_alias=AliasChoices("INTERNAL_FILES_URL", "SERVER_CONSOLE_API_URL"),
alias_priority=1,
" Falls back to FILES_URL if not specified.",
default="",
)
-16
View File
@@ -6,17 +6,6 @@ class PyProjectConfig(BaseModel):
version: str = Field(description="Dify version", default="")
class DifyToolConfig(BaseModel):
min_difyctl_version: str = Field(
description="Oldest difyctl version served on /openapi/v1",
default="0.0.0",
)
class ToolConfig(BaseModel):
dify: DifyToolConfig = Field(default=DifyToolConfig())
class PyProjectTomlConfig(BaseSettings):
"""
configs in api/pyproject.toml
@@ -26,8 +15,3 @@ class PyProjectTomlConfig(BaseSettings):
description="configs in the project section of pyproject.toml",
default=PyProjectConfig(),
)
tool: ToolConfig = Field(
description="configs in the [tool.*] section of pyproject.toml",
default=ToolConfig(),
)
+1 -6
View File
@@ -1,6 +1,6 @@
from __future__ import annotations
from typing import Any, Literal
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, RootModel, computed_field
@@ -52,11 +52,6 @@ class AudioTranscriptResponse(ResponseModel):
text: str
class ValidationResultResponse(ResponseModel):
result: Literal["success", "error"]
error: str | None = None
class SimpleResultMessageResponse(ResponseModel):
result: str
message: str
+2 -2
View File
@@ -707,7 +707,7 @@ class AppStarApi(Resource):
@with_session
@get_app_model(mode=None)
def post(self, session: Session, current_user_id: str, app_model: App):
AppService.star_app(app=app_model, account_id=current_user_id, session=session)
AppService.star_app(session, app=app_model, account_id=current_user_id)
return SimpleResultResponse(result="success").model_dump(mode="json")
@console_ns.doc("unstar_app")
@@ -723,7 +723,7 @@ class AppStarApi(Resource):
@with_session
@get_app_model(mode=None)
def delete(self, session: Session, current_user_id: str, app_model: App):
AppService.unstar_app(app=app_model, account_id=current_user_id, session=session)
AppService.unstar_app(session, app=app_model, account_id=current_user_id)
return SimpleResultResponse(result="success").model_dump(mode="json")
+1 -2
View File
@@ -36,7 +36,7 @@ from controllers.console.wraps import (
with_current_user_id,
)
from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
from core.app.entities.app_invoke_entities import AGENT_RUNTIME_EXIT_INTENT_ARG, InvokeFrom
from core.app.entities.app_invoke_entities import InvokeFrom
from core.app.features.rate_limiting.rate_limit import RateLimitGenerator
from core.errors.error import (
ModelCurrentlyNotSupportError,
@@ -416,7 +416,6 @@ def _create_build_chat_finalization_message(
"draft_type": "debug_build",
"conversation_id": debug_conversation_id,
"auto_generate_name": False,
AGENT_RUNTIME_EXIT_INTENT_ARG: "delete",
}
external_trace_id = get_external_trace_id(request)
if external_trace_id:
@@ -59,7 +59,7 @@ class ApiKeyAuthDataSource(Resource):
@account_initialization_required
@with_current_tenant_id
def get(self, current_tenant_id: str):
data_source_api_key_bindings = ApiKeyAuthService.get_provider_auth_list(current_tenant_id, session=db.session())
data_source_api_key_bindings = ApiKeyAuthService.get_provider_auth_list(db.session(), current_tenant_id)
if data_source_api_key_bindings:
return {
"sources": [
@@ -93,7 +93,7 @@ class ApiKeyAuthDataSourceBinding(Resource):
data = payload.model_dump()
ApiKeyAuthService.validate_api_key_auth_args(data)
try:
ApiKeyAuthService.create_provider_auth(current_tenant_id, data, session=db.session())
ApiKeyAuthService.create_provider_auth(db.session(), current_tenant_id, data)
except Exception as e:
raise ApiKeyAuthFailedError(str(e))
return {"result": "success"}, 200
@@ -110,6 +110,6 @@ class ApiKeyAuthDataSourceBindingDelete(Resource):
@with_current_tenant_id
def delete(self, current_tenant_id: str, binding_id: UUID):
# The role of the current user in the table must be admin or owner
ApiKeyAuthService.delete_provider_auth(current_tenant_id, str(binding_id), session=db.session())
ApiKeyAuthService.delete_provider_auth(db.session(), current_tenant_id, str(binding_id))
return "", 204
@@ -101,7 +101,7 @@ class EmailRegisterSendEmailApi(Resource):
if dify_config.BILLING_ENABLED and BillingService.is_email_in_freeze(normalized_email):
raise AccountInFreezeError()
account = AccountService.get_account_by_email_with_case_fallback(args.email, session=db.session())
account = AccountService.get_account_by_email_with_case_fallback(db.session(), args.email)
token = AccountService.send_email_register_email(email=normalized_email, account=account, language=language)
return {"result": "success", "data": token}
@@ -176,7 +176,7 @@ class EmailRegisterResetApi(Resource):
email = register_data.get("email", "")
normalized_email = email.lower()
account = AccountService.get_account_by_email_with_case_fallback(email, session=db.session())
account = AccountService.get_account_by_email_with_case_fallback(db.session(), email)
if account:
raise EmailAlreadyInUseError()
@@ -82,7 +82,7 @@ class ForgotPasswordSendEmailApi(Resource):
else:
language = "en-US"
account = AccountService.get_account_by_email_with_case_fallback(args.email, session=db.session())
account = AccountService.get_account_by_email_with_case_fallback(db.session(), args.email)
token = AccountService.send_reset_password_email(
account=account,
@@ -180,7 +180,7 @@ class ForgotPasswordResetApi(Resource):
password_hashed = hash_password(args.new_password, salt)
email = reset_data.get("email", "")
account = AccountService.get_account_by_email_with_case_fallback(email, session=db.session())
account = AccountService.get_account_by_email_with_case_fallback(db.session(), email)
if account:
account = db.session.merge(account)
+1 -1
View File
@@ -225,7 +225,7 @@ def _get_account_by_openid_or_email(provider: str, user_info: OAuthUserInfo) ->
account: Account | None = Account.get_by_openid(provider, user_info.id)
if not account:
account = AccountService.get_account_by_email_with_case_fallback(user_info.email, session=db.session())
account = AccountService.get_account_by_email_with_case_fallback(db.session(), user_info.email)
return account
+2 -2
View File
@@ -56,7 +56,7 @@ class Subscription(Resource):
@with_current_tenant_id
def get(self, current_tenant_id: str, current_user: Account):
args = SubscriptionQuery.model_validate(request.args.to_dict(flat=True))
BillingService.is_tenant_owner_or_admin(current_user, session=db.session())
BillingService.is_tenant_owner_or_admin(db.session(), current_user)
return BillingService.get_subscription(args.plan, args.interval, current_user.email, current_tenant_id)
@@ -70,7 +70,7 @@ class Invoices(Resource):
@with_current_user
@with_current_tenant_id
def get(self, current_tenant_id: str, current_user: Account):
BillingService.is_tenant_owner_or_admin(current_user, session=db.session())
BillingService.is_tenant_owner_or_admin(db.session(), current_user)
return BillingService.get_invoices(current_user.email, current_tenant_id)
+25 -25
View File
@@ -30,7 +30,6 @@ from controllers.console.wraps import (
with_current_tenant_id,
with_current_user,
)
from core.entities.knowledge_entities import IndexingEstimate
from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
from core.indexing_runner import IndexingRunner
from core.plugin.impl.model_runtime_factory import create_plugin_provider_manager
@@ -269,10 +268,21 @@ class ErrorDocsResponse(DocumentStatusListResponse):
total: int
class IndexingEstimateResponse(IndexingEstimate):
tokens: int
total_price: float | int
currency: str
class IndexingEstimatePreviewItemResponse(ResponseModel):
content: str
child_chunks: list[str] | None = None
summary: str | None = None
class IndexingEstimateQaPreviewItemResponse(ResponseModel):
question: str
answer: str
class IndexingEstimateResponse(ResponseModel):
total_segments: int
preview: list[IndexingEstimatePreviewItemResponse]
qa_preview: list[IndexingEstimateQaPreviewItemResponse] | None = None
class RetrievalSettingResponse(ResponseModel):
@@ -637,7 +647,7 @@ class DatasetApi(Resource):
else:
data["embedding_available"] = True
return dump_response(DatasetDetailWithPartialMembersResponse, data), 200
return data, 200
@console_ns.doc("update_dataset")
@console_ns.doc(description="Update dataset details")
@@ -678,10 +688,10 @@ class DatasetApi(Resource):
# The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
if not dify_config.RBAC_ENABLED:
DatasetPermissionService.check_permission(
current_user, dataset, payload.permission, payload.partial_member_list, session=session
session, current_user, dataset, payload.permission, payload.partial_member_list
)
dataset = DatasetService.update_dataset(dataset_id_str, payload_data, current_user, session=session)
dataset = DatasetService.update_dataset(session, dataset_id_str, payload_data, current_user)
if dataset is None:
raise NotFound("Dataset not found.")
@@ -707,7 +717,7 @@ class DatasetApi(Resource):
partial_member_list = DatasetPermissionService.get_dataset_partial_member_list(dataset_id_str, db.session())
result_data.update({"partial_member_list": partial_member_list})
return dump_response(DatasetDetailWithPartialMembersResponse, result_data), 200
return result_data, 200
@setup_required
@login_required
@@ -750,7 +760,7 @@ class DatasetUseCheckApi(Resource):
dataset_id_str = str(dataset_id)
dataset_is_using = DatasetService.dataset_use_check(dataset_id_str, db.session())
return UsageCheckResponse(is_using=dataset_is_using).model_dump(mode="json"), 200
return {"is_using": dataset_is_using}, 200
@console_ns.route("/datasets/<uuid:dataset_id>/queries")
@@ -891,17 +901,7 @@ class DatasetIndexingEstimateApi(Resource):
except Exception as e:
raise IndexingEstimateError(str(e))
return (
IndexingEstimateResponse(
tokens=0,
total_price=0,
currency="USD",
total_segments=response.total_segments,
preview=response.preview,
qa_preview=response.qa_preview,
).model_dump(mode="json", exclude_none=True),
200,
)
return response.model_dump(), 200
@console_ns.route("/datasets/<uuid:dataset_id>/related-apps")
@@ -1018,7 +1018,7 @@ class DatasetApiKeyApi(Resource):
keys = db.session.scalars(
select(ApiToken).where(ApiToken.type == self.resource_type, ApiToken.tenant_id == current_tenant_id)
).all()
return dump_response(ApiKeyList, {"data": keys})
return ApiKeyList.model_validate({"data": keys}, from_attributes=True).model_dump(mode="json")
@console_ns.response(200, "API key created successfully", console_ns.models[ApiKeyItem.__name__])
@console_ns.response(400, "Maximum keys exceeded")
@@ -1052,7 +1052,7 @@ class DatasetApiKeyApi(Resource):
api_token.type = self.resource_type
db.session.add(api_token)
db.session.commit()
return dump_response(ApiKeyItem, api_token), 200
return ApiKeyItem.model_validate(api_token, from_attributes=True).model_dump(mode="json"), 200
@console_ns.route("/datasets/api-keys/<uuid:api_key_id>")
@@ -1107,7 +1107,7 @@ class DatasetEnableApiApi(Resource):
DatasetService.update_dataset_api_status(dataset_id_str, status == "enable", db.session())
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
return {"result": "success"}, 200
@console_ns.route("/datasets/api-base-info")
@@ -1120,7 +1120,7 @@ class DatasetApiBaseUrlApi(Resource):
@account_initialization_required
def get(self):
base = dify_config.SERVICE_API_URL or request.host_url.rstrip("/")
return ApiBaseUrlResponse(api_base_url=normalize_api_base_url(base)).model_dump(mode="json")
return {"api_base_url": normalize_api_base_url(base)}
@console_ns.route("/datasets/retrieval-setting")
@@ -10,17 +10,16 @@ from uuid import UUID
import sqlalchemy as sa
from flask import request, send_file
from flask_restx import Resource
from pydantic import BaseModel, Field, JsonValue, field_validator
from pydantic import BaseModel, Field, RootModel, field_validator
from sqlalchemy import asc, desc, func, select
from werkzeug.exceptions import Forbidden, NotFound
import services
from controllers.common.controller_schemas import DocumentBatchDownloadZipPayload
from controllers.common.fields import SimpleResultMessageResponse, SimpleResultResponse, UrlResponse
from controllers.common.fields import BinaryFileResponse, SimpleResultMessageResponse, SimpleResultResponse, UrlResponse
from controllers.common.schema import register_response_schema_models, register_schema_models
from controllers.console import console_ns
from controllers.console.wraps import RBACPermission, RBACResourceScope, rbac_permission_required
from core.entities.knowledge_entities import IndexingEstimate
from core.errors.error import (
LLMBadRequestError,
ModelCurrentlyNotSupportError,
@@ -30,7 +29,6 @@ from core.errors.error import (
from core.indexing_runner import IndexingRunner
from core.model_manager import ModelManager
from core.plugin.impl.exc import PluginDaemonClientSideError
from core.rag.entities import Rule
from core.rag.extractor.entity.datasource_type import DatasourceType
from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo, WebsiteInfo
from core.rag.index_processor.constant.index_type import IndexTechniqueType
@@ -51,7 +49,7 @@ from libs.login import login_required
from libs.pagination import paginate_query
from models import Account, DatasetProcessRule, Document, DocumentSegment, UploadFile
from models.dataset import DocumentPipelineExecutionLog
from models.enums import IndexingStatus, ProcessRuleMode, SegmentStatus
from models.enums import IndexingStatus, SegmentStatus
from services.dataset_ref_service import DatasetRefService
from services.dataset_service import DatasetService, DocumentService
from services.entities.knowledge_entities.knowledge_entities import KnowledgeConfig, ProcessRule, RetrievalModel
@@ -150,91 +148,8 @@ class DocumentWithSegmentsListResponse(ResponseModel):
page: int
class IndexingEstimateResponse(IndexingEstimate):
tokens: int
total_price: float | int
currency: str
class DocumentDetailResponse(ResponseModel):
id: str
position: int | None = None
data_source_type: str | None = None
data_source_info: Any = None
data_source_detail_dict: Any = None
dataset_process_rule_id: str | None = None
dataset_process_rule: Any = None
document_process_rule: Any = None
name: str | None = None
created_from: str | None = None
created_by: str | None = None
created_at: int | None = None
tokens: int | None = None
indexing_status: str | None = None
completed_at: int | None = None
updated_at: int | None = None
indexing_latency: float | None = None
error: str | None = None
enabled: bool | None = None
disabled_at: int | None = None
disabled_by: str | None = None
archived: bool | None = None
doc_type: str | None = None
doc_metadata: list[DocumentMetadataResponse] | None = None
segment_count: int | None = None
average_segment_length: float | None = None
hit_count: int | None = None
display_status: str | None = None
doc_form: str | None = None
doc_language: str | None = None
need_summary: bool | None = None
@field_validator("data_source_type", "indexing_status", "display_status", "doc_form", mode="before")
@classmethod
def _normalize_enum_fields(cls, value: Any) -> Any:
return normalize_enum(value)
class SummaryStatusResponse(ResponseModel):
completed: int = 0
generating: int = 0
error: int = 0
not_started: int = 0
timeout: int = 0
class SummaryEntryResponse(ResponseModel):
segment_id: str
segment_position: int
status: str
summary_preview: str | None = None
error: str | None = None
created_at: int | None = None
updated_at: int | None = None
@field_validator("status", mode="before")
@classmethod
def _normalize_status(cls, value: Any) -> Any:
return normalize_enum(value)
class DocumentSummaryStatusResponse(ResponseModel):
total_segments: int
summary_status: SummaryStatusResponse
summaries: list[SummaryEntryResponse]
class ProcessRuleResponse(ResponseModel):
mode: ProcessRuleMode
rules: Rule | None = None
limits: dict[str, Any]
class DocumentPipelineExecutionLogResponse(ResponseModel):
datasource_info: JsonValue | None = None
datasource_type: str | None = None
input_data: JsonValue | None = None
datasource_node_id: str | None = None
class OpaqueObjectResponse(RootModel[dict[str, Any]]):
root: dict[str, Any]
register_schema_models(
@@ -250,6 +165,7 @@ register_schema_models(
)
register_response_schema_models(
console_ns,
BinaryFileResponse,
SimpleResultMessageResponse,
SimpleResultResponse,
UrlResponse,
@@ -259,11 +175,7 @@ register_response_schema_models(
DocumentWithSegmentsResponse,
DatasetAndDocumentResponse,
DocumentWithSegmentsListResponse,
IndexingEstimateResponse,
DocumentDetailResponse,
DocumentSummaryStatusResponse,
ProcessRuleResponse,
DocumentPipelineExecutionLogResponse,
OpaqueObjectResponse,
)
@@ -313,7 +225,7 @@ class GetProcessRuleApi(Resource):
@console_ns.doc("get_process_rule")
@console_ns.doc(description="Get dataset document processing rules")
@console_ns.doc(params={"document_id": "Document ID (optional)"})
@console_ns.response(200, "Process rules retrieved successfully", console_ns.models[ProcessRuleResponse.__name__])
@console_ns.response(200, "Process rules retrieved successfully", console_ns.models[OpaqueObjectResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -352,7 +264,7 @@ class GetProcessRuleApi(Resource):
mode = dataset_process_rule.mode
rules = dataset_process_rule.rules_dict
return dump_response(ProcessRuleResponse, {"mode": mode, "rules": rules, "limits": limits})
return {"mode": mode, "rules": rules, "limits": limits}
@console_ns.route("/datasets/<uuid:dataset_id>/documents")
@@ -579,7 +491,7 @@ class DatasetInitApi(Resource):
@console_ns.doc(description="Initialize dataset with documents")
@console_ns.expect(console_ns.models[KnowledgeConfig.__name__])
@console_ns.response(
200, "Dataset initialized successfully", console_ns.models[DatasetAndDocumentResponse.__name__]
201, "Dataset initialized successfully", console_ns.models[DatasetAndDocumentResponse.__name__]
)
@console_ns.response(400, "Invalid request parameters")
@setup_required
@@ -645,7 +557,7 @@ class DocumentIndexingEstimateApi(DocumentResource):
@console_ns.response(
200,
"Indexing estimate calculated successfully",
console_ns.models[IndexingEstimateResponse.__name__],
console_ns.models[OpaqueObjectResponse.__name__],
)
@console_ns.response(404, "Document not found")
@console_ns.response(400, "Document already finished")
@@ -666,6 +578,8 @@ class DocumentIndexingEstimateApi(DocumentResource):
data_process_rule = document.dataset_process_rule
data_process_rule_dict = data_process_rule.to_dict() if data_process_rule else {}
response = {"tokens": 0, "total_price": 0, "currency": "USD", "total_segments": 0, "preview": []}
if document.data_source_type == "upload_file":
data_source_info = document.data_source_info_dict
if data_source_info and "upload_file_id" in data_source_info:
@@ -696,18 +610,7 @@ class DocumentIndexingEstimateApi(DocumentResource):
"English",
dataset_id_str,
)
return (
# TODO: why using zero here? the same for the below endpoint
IndexingEstimateResponse(
tokens=0,
total_price=0,
currency="USD",
total_segments=estimate_response.total_segments,
preview=estimate_response.preview,
qa_preview=estimate_response.qa_preview,
).model_dump(mode="json", exclude_none=True),
200,
)
return estimate_response.model_dump(), 200
except LLMBadRequestError:
raise ProviderNotInitializeError(
"No Embedding Model available. Please configure a valid provider "
@@ -720,24 +623,15 @@ class DocumentIndexingEstimateApi(DocumentResource):
except Exception as e:
raise IndexingEstimateError(str(e))
return (
IndexingEstimateResponse(
tokens=0,
total_price=0,
currency="USD",
total_segments=0,
preview=[],
).model_dump(mode="json", exclude_none=True),
200,
)
return response, 200
@console_ns.route("/datasets/<uuid:dataset_id>/batch/<string:batch>/indexing-estimate")
class DocumentBatchIndexingEstimateApi(DocumentResource):
@console_ns.response(
200,
"Indexing estimate calculated successfully",
console_ns.models[IndexingEstimateResponse.__name__],
"Batch indexing estimate calculated successfully",
console_ns.models[OpaqueObjectResponse.__name__],
)
@setup_required
@login_required
@@ -749,16 +643,7 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
dataset_id_str = str(dataset_id)
documents = self.get_batch_documents(dataset_id_str, batch, current_user)
if not documents:
return (
IndexingEstimateResponse(
tokens=0,
total_price=0,
currency="USD",
total_segments=0,
preview=[],
).model_dump(mode="json", exclude_none=True),
200,
)
return {"tokens": 0, "total_price": 0, "currency": "USD", "total_segments": 0, "preview": []}, 200
data_process_rule = documents[0].dataset_process_rule
data_process_rule_dict = data_process_rule.to_dict() if data_process_rule else {}
extract_settings = []
@@ -832,17 +717,7 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
"English",
dataset_id_str,
)
return (
IndexingEstimateResponse(
tokens=0,
total_price=0,
currency="USD",
total_segments=response.total_segments,
preview=response.preview,
qa_preview=response.qa_preview,
).model_dump(mode="json", exclude_none=True),
200,
)
return response.model_dump(), 200
except LLMBadRequestError:
raise ProviderNotInitializeError(
"No Embedding Model available. Please configure a valid provider in the Settings -> Model Provider."
@@ -979,7 +854,7 @@ class DocumentApi(DocumentResource):
"metadata": "Metadata inclusion (all/only/without)",
}
)
@console_ns.response(200, "Document retrieved successfully", console_ns.models[DocumentDetailResponse.__name__])
@console_ns.response(200, "Document retrieved successfully", console_ns.models[OpaqueObjectResponse.__name__])
@console_ns.response(404, "Document not found")
@setup_required
@login_required
@@ -996,21 +871,46 @@ class DocumentApi(DocumentResource):
if metadata not in self.METADATA_CHOICES:
raise InvalidMetadataError(f"Invalid metadata value: {metadata}")
metadata_fields = {"doc_type", "doc_metadata"}
if metadata == "only":
response = DocumentDetailResponse.model_validate(
{
"id": document.id,
"doc_type": document.doc_type,
"doc_metadata": document.doc_metadata_details,
}
)
return response.model_dump(mode="json", include={"id", *metadata_fields}, exclude_unset=True), 200
dataset_process_rules = DatasetService.get_process_rules(dataset_id_str, db.session())
document_process_rules = document.dataset_process_rule.to_dict() if document.dataset_process_rule else {}
response = DocumentDetailResponse.model_validate(
{
response = {"id": document.id, "doc_type": document.doc_type, "doc_metadata": document.doc_metadata_details}
elif metadata == "without":
dataset_process_rules = DatasetService.get_process_rules(dataset_id_str, db.session())
document_process_rules = document.dataset_process_rule.to_dict() if document.dataset_process_rule else {}
response = {
"id": document.id,
"position": document.position,
"data_source_type": document.data_source_type,
"data_source_info": document.data_source_info_dict,
"data_source_detail_dict": document.data_source_detail_dict,
"dataset_process_rule_id": document.dataset_process_rule_id,
"dataset_process_rule": dataset_process_rules,
"document_process_rule": document_process_rules,
"name": document.name,
"created_from": document.created_from,
"created_by": document.created_by,
"created_at": int(document.created_at.timestamp()),
"tokens": document.tokens,
"indexing_status": document.indexing_status,
"completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,
"updated_at": int(document.updated_at.timestamp()) if document.updated_at else None,
"indexing_latency": document.indexing_latency,
"error": document.error,
"enabled": document.enabled,
"disabled_at": int(document.disabled_at.timestamp()) if document.disabled_at else None,
"disabled_by": document.disabled_by,
"archived": document.archived,
"segment_count": document.segment_count,
"average_segment_length": document.average_segment_length,
"hit_count": document.hit_count,
"display_status": document.display_status,
"doc_form": document.doc_form,
"doc_language": document.doc_language,
"need_summary": document.need_summary if document.need_summary is not None else False,
}
else:
dataset_process_rules = DatasetService.get_process_rules(dataset_id_str, db.session())
document_process_rules = document.dataset_process_rule.to_dict() if document.dataset_process_rule else {}
response = {
"id": document.id,
"position": document.position,
"data_source_type": document.data_source_type,
@@ -1043,9 +943,8 @@ class DocumentApi(DocumentResource):
"doc_language": document.doc_language,
"need_summary": document.need_summary if document.need_summary is not None else False,
}
)
exclude = metadata_fields if metadata == "without" else None
return response.model_dump(mode="json", exclude=exclude, exclude_unset=True), 200
return response, 200
@setup_required
@login_required
@@ -1091,9 +990,7 @@ class DocumentDownloadApi(DocumentResource):
def get(self, current_tenant_id: str, current_user: Account, dataset_id: UUID, document_id: UUID) -> dict[str, Any]:
# Reuse the shared permission/tenant checks implemented in DocumentResource.
document = self.get_document(str(dataset_id), str(document_id), current_user, current_tenant_id)
return UrlResponse(url=DocumentService.get_document_download_url(document, db.session())).model_dump(
mode="json"
)
return {"url": DocumentService.get_document_download_url(document, db.session())}
@console_ns.route("/datasets/<uuid:dataset_id>/documents/download-zip")
@@ -1102,7 +999,7 @@ class DocumentBatchDownloadZipApi(DocumentResource):
@console_ns.doc("download_dataset_documents_as_zip")
@console_ns.doc(description="Download selected dataset documents as a single ZIP archive (upload-file only)")
@console_ns.response(200, "ZIP archive downloaded successfully")
@console_ns.response(200, "ZIP archive generated successfully", console_ns.models[BinaryFileResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -1137,7 +1034,6 @@ class DocumentBatchDownloadZipApi(DocumentResource):
)
cleanup = stack.pop_all()
response.call_on_close(cleanup.close)
# response-contract:ignore binary ZIP download response
return response
@@ -1197,7 +1093,7 @@ class DocumentProcessingApi(DocumentResource):
document.is_paused = False
db.session.commit()
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
return {"result": "success"}, 200
@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/metadata")
@@ -1256,9 +1152,7 @@ class DocumentMetadataApi(DocumentResource):
document.updated_at = naive_utc_now()
db.session.commit()
return SimpleResultMessageResponse(result="success", message="Document metadata updated.").model_dump(
mode="json"
), 200
return {"result": "success", "message": "Document metadata updated."}, 200
@console_ns.route("/datasets/<uuid:dataset_id>/documents/status/<string:action>/batch")
@@ -1300,7 +1194,7 @@ class DocumentStatusApi(DocumentResource):
except NotFound as e:
raise NotFound(str(e))
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
return {"result": "success"}, 200
@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/processing/pause")
@@ -1427,7 +1321,7 @@ class DocumentRenameApi(DocumentResource):
# The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
if not current_user.is_dataset_editor:
raise Forbidden()
dataset = DatasetService.get_dataset(str(dataset_id), db.session())
dataset = DatasetService.get_dataset(dataset_id, db.session())
if not dataset:
raise NotFound("Dataset not found.")
DatasetService.check_dataset_operator_permission(current_user, dataset, session=db.session())
@@ -1469,15 +1363,15 @@ class WebsiteDocumentSyncApi(DocumentResource):
# sync document
DocumentService.sync_website_document(dataset_id_str, document, db.session())
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
return {"result": "success"}, 200
@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/pipeline-execution-log")
class DocumentPipelineExecutionLogApi(DocumentResource):
@console_ns.response(
200,
"Pipeline execution log retrieved successfully",
console_ns.models[DocumentPipelineExecutionLogResponse.__name__],
"Document pipeline execution log retrieved successfully",
console_ns.models[OpaqueObjectResponse.__name__],
)
@setup_required
@login_required
@@ -1500,16 +1394,18 @@ class DocumentPipelineExecutionLogApi(DocumentResource):
.limit(1)
)
if not log:
return DocumentPipelineExecutionLogResponse().model_dump(mode="json"), 200
return dump_response(
DocumentPipelineExecutionLogResponse,
{
"datasource_info": json.loads(log.datasource_info),
"datasource_type": log.datasource_type,
"input_data": log.input_data,
"datasource_node_id": log.datasource_node_id,
},
), 200
return {
"datasource_info": None,
"datasource_type": None,
"input_data": None,
"datasource_node_id": None,
}, 200
return {
"datasource_info": json.loads(log.datasource_info),
"datasource_type": log.datasource_type,
"input_data": log.input_data,
"datasource_node_id": log.datasource_node_id,
}, 200
@console_ns.route("/datasets/<uuid:dataset_id>/documents/generate-summary")
@@ -1612,7 +1508,7 @@ class DocumentGenerateSummaryApi(Resource):
dataset_id_str,
)
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
return {"result": "success"}, 200
@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/summary-status")
@@ -1620,11 +1516,7 @@ class DocumentSummaryStatusApi(DocumentResource):
@console_ns.doc("get_document_summary_status")
@console_ns.doc(description="Get summary index generation status for a document")
@console_ns.doc(params={"dataset_id": "Dataset ID", "document_id": "Document ID"})
@console_ns.response(
200,
"Summary status retrieved successfully",
console_ns.models[DocumentSummaryStatusResponse.__name__],
)
@console_ns.response(200, "Summary status retrieved successfully", console_ns.models[OpaqueObjectResponse.__name__])
@console_ns.response(404, "Document not found")
@setup_required
@login_required
@@ -1642,7 +1534,6 @@ class DocumentSummaryStatusApi(DocumentResource):
- generating: Number of summaries being generated
- error: Number of summaries with errors
- not_started: Number of segments without summary records
- timeout: Number of summaries that timed out
- summaries: List of summary records with status and content preview
"""
dataset_id_str = str(dataset_id)
@@ -1668,4 +1559,4 @@ class DocumentSummaryStatusApi(DocumentResource):
session=db.session(),
)
return dump_response(DocumentSummaryStatusResponse, result), 200
return result, 200
@@ -391,7 +391,7 @@ class DatasetDocumentSegmentApi(Resource):
SegmentService.update_segments_status(segment_ids, action, dataset, document, db.session())
except Exception as e:
raise InvalidActionError(str(e))
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
return dump_response(SimpleResultResponse, {"result": "success"}), 200
@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segment")
+107 -89
View File
@@ -1,16 +1,20 @@
from datetime import datetime
from typing import Any
from uuid import UUID
from flask import request
from flask_restx import Resource
from pydantic import AliasChoices, BaseModel, Field, field_validator
from flask_restx import Resource, fields, marshal
from pydantic import BaseModel, Field, RootModel
from sqlalchemy.orm import Session
from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
import services
from controllers.common.fields import UsageCountResponse
from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models
from controllers.common.schema import (
get_or_create_model,
query_params_from_model,
register_response_schema_models,
register_schema_models,
)
from controllers.console import console_ns
from controllers.console.app.wraps import with_session
from controllers.console.datasets.error import DatasetNameDuplicateError
@@ -24,9 +28,21 @@ from controllers.console.wraps import (
with_current_tenant_id,
with_current_user,
)
from extensions.ext_database import db
from fields.base import ResponseModel
from fields.dataset_fields import DatasetDetailResponse
from libs.helper import dump_response
from fields.dataset_fields import (
dataset_detail_fields,
dataset_retrieval_model_fields,
doc_metadata_fields,
external_knowledge_info_fields,
external_retrieval_model_fields,
icon_info_fields,
keyword_setting_fields,
reranking_model_fields,
tag_fields,
vector_setting_fields,
weighted_score_fields,
)
from libs.login import login_required
from models import Account
from services.dataset_service import DatasetService
@@ -35,10 +51,50 @@ from services.external_knowledge_service import ExternalDatasetService
from services.hit_testing_service import HitTestingService
from services.knowledge_service import BedrockRetrievalSetting, ExternalDatasetTestService
register_response_schema_models(console_ns, UsageCountResponse)
def _build_dataset_detail_model():
keyword_setting_model = get_or_create_model("DatasetKeywordSetting", keyword_setting_fields)
vector_setting_model = get_or_create_model("DatasetVectorSetting", vector_setting_fields)
weighted_score_fields_copy = weighted_score_fields.copy()
weighted_score_fields_copy["keyword_setting"] = fields.Nested(keyword_setting_model)
weighted_score_fields_copy["vector_setting"] = fields.Nested(vector_setting_model)
weighted_score_model = get_or_create_model("DatasetWeightedScore", weighted_score_fields_copy)
reranking_model = get_or_create_model("DatasetRerankingModel", reranking_model_fields)
dataset_retrieval_model_fields_copy = dataset_retrieval_model_fields.copy()
dataset_retrieval_model_fields_copy["reranking_model"] = fields.Nested(reranking_model)
dataset_retrieval_model_fields_copy["weights"] = fields.Nested(weighted_score_model, allow_null=True)
dataset_retrieval_model = get_or_create_model("DatasetRetrievalModel", dataset_retrieval_model_fields_copy)
tag_model = get_or_create_model("Tag", tag_fields)
doc_metadata_model = get_or_create_model("DatasetDocMetadata", doc_metadata_fields)
external_knowledge_info_model = get_or_create_model("ExternalKnowledgeInfo", external_knowledge_info_fields)
external_retrieval_model = get_or_create_model("ExternalRetrievalModel", external_retrieval_model_fields)
icon_info_model = get_or_create_model("DatasetIconInfo", icon_info_fields)
dataset_detail_fields_copy = dataset_detail_fields.copy()
dataset_detail_fields_copy["retrieval_model_dict"] = fields.Nested(dataset_retrieval_model)
dataset_detail_fields_copy["tags"] = fields.List(fields.Nested(tag_model))
dataset_detail_fields_copy["external_knowledge_info"] = fields.Nested(external_knowledge_info_model)
dataset_detail_fields_copy["external_retrieval_model"] = fields.Nested(external_retrieval_model, allow_null=True)
dataset_detail_fields_copy["doc_metadata"] = fields.List(fields.Nested(doc_metadata_model))
dataset_detail_fields_copy["icon_info"] = fields.Nested(icon_info_model)
return get_or_create_model("DatasetDetail", dataset_detail_fields_copy)
try:
dataset_detail_model = console_ns.models["DatasetDetail"]
except KeyError:
dataset_detail_model = _build_dataset_detail_model()
class ExternalKnowledgeApiPayload(BaseModel):
name: str = Field(..., min_length=1, max_length=40)
settings: dict[str, Any]
settings: dict[str, object]
class ExternalDatasetCreatePayload(BaseModel):
@@ -46,13 +102,15 @@ class ExternalDatasetCreatePayload(BaseModel):
external_knowledge_id: str
name: str = Field(..., min_length=1, max_length=100)
description: str | None = Field(None, max_length=400)
external_retrieval_model: dict[str, Any] | None = None
external_retrieval_model: dict[str, object] | None = Field(default=None)
class ExternalHitTestingPayload(BaseModel):
query: str
external_retrieval_model: dict[str, Any] | None = None
metadata_filtering_conditions: dict[str, Any] | None = None
external_retrieval_model: dict[str, object] | None = Field(default=None)
metadata_filtering_conditions: dict[str, object] | None = Field(
default=None,
)
class BedrockRetrievalPayload(BaseModel):
@@ -67,7 +125,7 @@ class ExternalApiTemplateListQuery(BaseModel):
keyword: str | None = Field(default=None, description="Search keyword")
class ExternalKnowledgeApiBindingResponse(ResponseModel):
class ExternalKnowledgeDatasetBindingResponse(ResponseModel):
id: str
name: str
@@ -77,52 +135,22 @@ class ExternalKnowledgeApiResponse(ResponseModel):
tenant_id: str
name: str
description: str
settings: dict[str, Any] | None = Field(validation_alias=AliasChoices("settings_dict", "settings"))
dataset_bindings: list[ExternalKnowledgeApiBindingResponse]
settings: dict[str, Any] | None = Field(default=None)
dataset_bindings: list[ExternalKnowledgeDatasetBindingResponse] = Field(default_factory=list)
created_by: str
created_at: str
@field_validator("created_at", mode="before")
@classmethod
def _normalize_created_at(cls, value: datetime | str) -> str:
if isinstance(value, datetime):
return value.isoformat()
return value
class ExternalKnowledgeApiListResponse(ResponseModel):
data: list[ExternalKnowledgeApiResponse]
has_more: bool
limit: int
total: int | None
total: int
page: int
class ExternalHitTestingQueryResponse(ResponseModel):
content: str
class ExternalHitTestingRecordResponse(ResponseModel):
content: str | None = None
title: str | None = None
score: float | None = None
metadata: dict[str, Any] | None = None
class ExternalHitTestingResponse(ResponseModel):
query: ExternalHitTestingQueryResponse
records: list[ExternalHitTestingRecordResponse]
class BedrockRetrievalRecordResponse(ResponseModel):
metadata: dict[str, Any] | None = None
score: float
title: str | None = None
content: str | None = None
class BedrockRetrievalResponse(ResponseModel):
records: list[BedrockRetrievalRecordResponse]
class ExternalRetrievalTestResponse(RootModel[dict[str, Any] | list[dict[str, Any]]]):
root: dict[str, Any] | list[dict[str, Any]]
register_schema_models(
@@ -135,16 +163,9 @@ register_schema_models(
)
register_response_schema_models(
console_ns,
UsageCountResponse,
DatasetDetailResponse,
ExternalKnowledgeApiBindingResponse,
ExternalKnowledgeApiResponse,
ExternalKnowledgeApiListResponse,
ExternalHitTestingQueryResponse,
ExternalHitTestingRecordResponse,
ExternalHitTestingResponse,
BedrockRetrievalRecordResponse,
BedrockRetrievalResponse,
ExternalRetrievalTestResponse,
)
@@ -168,26 +189,24 @@ class ExternalApiTemplateListApi(Resource):
external_knowledge_apis, total = ExternalDatasetService.get_external_knowledge_apis(
query.page, query.limit, current_tenant_id, query.keyword
)
return ExternalKnowledgeApiListResponse(
data=[ExternalKnowledgeApiResponse.model_validate(item) for item in external_knowledge_apis],
has_more=len(external_knowledge_apis) == query.limit,
limit=query.limit,
total=total,
page=query.page,
).model_dump(mode="json"), 200
response = {
"data": [item.to_dict() for item in external_knowledge_apis],
"has_more": len(external_knowledge_apis) == query.limit,
"limit": query.limit,
"total": total,
"page": query.page,
}
return response, 200
@console_ns.doc("create_external_api_template")
@console_ns.doc(description="Create external knowledge API template")
@setup_required
@login_required
@account_initialization_required
@console_ns.expect(console_ns.models[ExternalKnowledgeApiPayload.__name__])
@console_ns.response(
201,
"External API template created successfully",
console_ns.models[ExternalKnowledgeApiResponse.__name__],
)
@console_ns.response(403, "Permission denied")
@setup_required
@login_required
@account_initialization_required
@with_current_user
@with_current_tenant_id
@with_session
@@ -210,7 +229,7 @@ class ExternalApiTemplateListApi(Resource):
except services.errors.dataset.DatasetNameDuplicateError:
raise DatasetNameDuplicateError()
return dump_response(ExternalKnowledgeApiResponse, external_knowledge_api), 201
return external_knowledge_api.to_dict(), 201
@console_ns.route("/datasets/external-knowledge-api/<uuid:external_knowledge_api_id>")
@@ -237,21 +256,17 @@ class ExternalApiTemplateApi(Resource):
if external_knowledge_api is None:
raise NotFound("API template not found.")
return dump_response(ExternalKnowledgeApiResponse, external_knowledge_api), 200
return external_knowledge_api.to_dict(), 200
@console_ns.doc("update_external_api_template")
@console_ns.doc(description="Update external knowledge API template")
@console_ns.doc(params={"external_knowledge_api_id": "External knowledge API ID"})
@console_ns.expect(console_ns.models[ExternalKnowledgeApiPayload.__name__])
@console_ns.response(
200,
"External API template updated successfully",
console_ns.models[ExternalKnowledgeApiResponse.__name__],
)
@console_ns.response(404, "Template not found")
@setup_required
@login_required
@account_initialization_required
@console_ns.expect(console_ns.models[ExternalKnowledgeApiPayload.__name__])
@with_current_user
@with_current_tenant_id
@with_session
@@ -269,7 +284,7 @@ class ExternalApiTemplateApi(Resource):
session=session,
)
return dump_response(ExternalKnowledgeApiResponse, external_knowledge_api), 200
return external_knowledge_api.to_dict(), 200
@setup_required
@login_required
@@ -307,7 +322,7 @@ class ExternalApiUseCheckApi(Resource):
external_knowledge_api_is_using, count = ExternalDatasetService.external_knowledge_api_use_check(
external_knowledge_api_id_str, current_tenant_id, session=session
)
return UsageCountResponse(is_using=external_knowledge_api_is_using, count=count).model_dump(mode="json"), 200
return {"is_using": external_knowledge_api_is_using, "count": count}, 200
@console_ns.route("/datasets/external")
@@ -315,9 +330,7 @@ class ExternalDatasetCreateApi(Resource):
@console_ns.doc("create_external_dataset")
@console_ns.doc(description="Create external knowledge dataset")
@console_ns.expect(console_ns.models[ExternalDatasetCreatePayload.__name__])
@console_ns.response(
201, "External dataset created successfully", console_ns.models[DatasetDetailResponse.__name__]
)
@console_ns.response(201, "External dataset created successfully", dataset_detail_model)
@console_ns.response(400, "Invalid parameters")
@console_ns.response(403, "Permission denied")
@setup_required
@@ -347,16 +360,17 @@ class ExternalDatasetCreateApi(Resource):
except services.errors.dataset.DatasetNameDuplicateError:
raise DatasetNameDuplicateError()
dataset_id_str = str(dataset.id)
item = marshal(dataset, dataset_detail_fields)
dataset_id_str = item["id"]
permission_keys_map = enterprise_rbac_service.RBACService.DatasetPermissions.batch_get(
str(current_tenant_id),
current_user.id,
[dataset_id_str],
session=session,
)
data = DatasetDetailResponse.model_validate(dataset).model_dump(mode="json")
data["permission_keys"] = permission_keys_map.get(dataset_id_str, [])
return data, 201
item["permission_keys"] = permission_keys_map.get(dataset_id_str, [])
return item, 201
@console_ns.route("/datasets/<uuid:dataset_id>/external-hit-testing")
@@ -368,7 +382,7 @@ class ExternalKnowledgeHitTestingApi(Resource):
@console_ns.response(
200,
"External hit testing completed successfully",
console_ns.models[ExternalHitTestingResponse.__name__],
console_ns.models[ExternalRetrievalTestResponse.__name__],
)
@console_ns.response(404, "Dataset not found")
@console_ns.response(400, "Invalid parameters")
@@ -380,12 +394,12 @@ class ExternalKnowledgeHitTestingApi(Resource):
@with_session
def post(self, session: Session, current_user: Account, dataset_id: UUID):
dataset_id_str = str(dataset_id)
dataset = DatasetService.get_dataset(dataset_id_str, session)
dataset = DatasetService.get_dataset(dataset_id_str, db.session())
if dataset is None:
raise NotFound("Dataset not found.")
try:
DatasetService.check_dataset_permission(dataset, current_user, session)
DatasetService.check_dataset_permission(dataset, current_user, db.session())
except services.errors.account.NoPermissionError as e:
raise Forbidden(str(e))
@@ -402,7 +416,7 @@ class ExternalKnowledgeHitTestingApi(Resource):
metadata_filtering_conditions=payload.metadata_filtering_conditions,
)
return dump_response(ExternalHitTestingResponse, response)
return response
except Exception as e:
raise InternalServerError(str(e))
@@ -413,7 +427,11 @@ class BedrockRetrievalApi(Resource):
@console_ns.doc("bedrock_retrieval_test")
@console_ns.doc(description="Bedrock retrieval test (internal use only)")
@console_ns.expect(console_ns.models[BedrockRetrievalPayload.__name__])
@console_ns.response(200, "Bedrock retrieval test completed", console_ns.models[BedrockRetrievalResponse.__name__])
@console_ns.response(
200,
"Bedrock retrieval test completed",
console_ns.models[ExternalRetrievalTestResponse.__name__],
)
def post(self):
payload = BedrockRetrievalPayload.model_validate(console_ns.payload or {})
@@ -421,4 +439,4 @@ class BedrockRetrievalApi(Resource):
result = ExternalDatasetTestService.knowledge_retrieval(
payload.retrieval_setting, payload.query, payload.knowledge_id
)
return dump_response(BedrockRetrievalResponse, result), 200
return result, 200
+7 -7
View File
@@ -67,7 +67,7 @@ class DatasetMetadataCreateApi(Resource):
DatasetService.check_dataset_permission(dataset, current_user, db.session())
metadata = MetadataService.create_metadata(
dataset_id_str, metadata_args, current_user, current_tenant_id, session=db.session()
db.session(), dataset_id_str, metadata_args, current_user, current_tenant_id
)
return dump_response(DatasetMetadataResponse, metadata), 201
@@ -84,7 +84,7 @@ class DatasetMetadataCreateApi(Resource):
dataset = DatasetService.get_dataset(dataset_id_str, db.session())
if dataset is None:
raise NotFound("Dataset not found.")
metadata = MetadataService.get_dataset_metadatas(dataset, session=db.session())
metadata = MetadataService.get_dataset_metadatas(db.session(), dataset)
return dump_response(DatasetMetadataListResponse, metadata), 200
@@ -111,7 +111,7 @@ class DatasetMetadataApi(Resource):
DatasetService.check_dataset_permission(dataset, current_user, db.session())
metadata = MetadataService.update_metadata_name(
dataset_id_str, metadata_id_str, name, current_user, current_tenant_id, session=db.session()
db.session(), dataset_id_str, metadata_id_str, name, current_user, current_tenant_id
)
return dump_response(DatasetMetadataResponse, metadata), 200
@@ -130,7 +130,7 @@ class DatasetMetadataApi(Resource):
raise NotFound("Dataset not found.")
DatasetService.check_dataset_permission(dataset, current_user, db.session())
MetadataService.delete_metadata(dataset_id_str, metadata_id_str, session=db.session())
MetadataService.delete_metadata(db.session(), dataset_id_str, metadata_id_str)
# Frontend callers only await success and invalidate metadata caches; no response body is consumed.
return "", 204
@@ -169,9 +169,9 @@ class DatasetMetadataBuiltInFieldActionApi(Resource):
match action:
case "enable":
MetadataService.enable_built_in_field(dataset, session=db.session())
MetadataService.enable_built_in_field(db.session(), dataset)
case "disable":
MetadataService.disable_built_in_field(dataset, session=db.session())
MetadataService.disable_built_in_field(db.session(), dataset)
# Frontend callers only await success and invalidate metadata caches; no response body is consumed.
return "", 204
@@ -198,7 +198,7 @@ class DocumentMetadataEditApi(Resource):
metadata_args = MetadataOperationData.model_validate(console_ns.payload or {})
MetadataService.update_documents_metadata(dataset, metadata_args, current_user, session=db.session())
MetadataService.update_documents_metadata(db.session(), dataset, metadata_args, current_user)
# Frontend callers only await success and invalidate caches; no response body is consumed.
return "", 204
@@ -343,8 +343,8 @@ class DraftRagPipelineRunApi(Resource):
@edit_permission_required
@rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_EDIT)
@with_current_user
@with_session
@get_rag_pipeline
@with_session
def post(self, session: Session, current_user: Account, pipeline: Pipeline):
"""
Run draft workflow
@@ -377,8 +377,8 @@ class PublishedRagPipelineRunApi(Resource):
@edit_permission_required
@rbac_permission_required(RBACResourceScope.DATASET, RBACPermission.DATASET_EDIT)
@with_current_user
@with_session
@get_rag_pipeline
@with_session
def post(self, session: Session, current_user: Account, pipeline: Pipeline):
"""
Run published workflow
+3 -5
View File
@@ -2,7 +2,6 @@ from collections.abc import Callable
from functools import wraps
from sqlalchemy import select
from sqlalchemy.orm import Session
from controllers.console.datasets.error import PipelineNotFoundError
from extensions.ext_database import db
@@ -23,10 +22,9 @@ def get_rag_pipeline[**P, R](view_func: Callable[P, R]) -> Callable[P, R]:
del kwargs["pipeline_id"]
stmt = select(Pipeline).where(Pipeline.id == pipeline_id, Pipeline.tenant_id == current_tenant_id).limit(1)
# Migrated handlers pass the request Session as args[1]; legacy handlers still use db.session.
session = args[1] if len(args) > 1 and isinstance(args[1], Session) else db.session
pipeline = session.scalar(stmt)
pipeline = db.session.scalar(
select(Pipeline).where(Pipeline.id == pipeline_id, Pipeline.tenant_id == current_tenant_id).limit(1)
)
if not pipeline:
raise PipelineNotFoundError()
@@ -120,7 +120,7 @@ class RecommendedAppListApi(Resource):
language_prefix = _resolve_language(args.language, current_user)
return RecommendedAppListResponse.model_validate(
RecommendedAppService.get_recommended_apps_and_categories(language_prefix, session=db.session()),
RecommendedAppService.get_recommended_apps_and_categories(db.session(), language_prefix),
from_attributes=True,
).model_dump(mode="json")
@@ -137,7 +137,7 @@ class LearnDifyAppListApi(Resource):
language_prefix = _resolve_language(args.language, current_user)
return LearnDifyAppListResponse.model_validate(
RecommendedAppService.get_learn_dify_apps(language_prefix, session=db.session()),
RecommendedAppService.get_learn_dify_apps(db.session(), language_prefix),
from_attributes=True,
).model_dump(mode="json")
@@ -148,4 +148,4 @@ class RecommendedAppApi(Resource):
@login_required
@account_initialization_required
def get(self, app_id: UUID):
return RecommendedAppService.get_recommend_app_detail(str(app_id), session=db.session())
return RecommendedAppService.get_recommend_app_detail(db.session(), str(app_id))
@@ -38,7 +38,11 @@ class SavedMessageListApi(InstalledAppResource):
args = SavedMessageListQuery.model_validate(request.args.to_dict())
pagination = SavedMessageService.pagination_by_last_id(
app_model, current_user, str(args.last_id) if args.last_id else None, args.limit, session=db.session()
db.session(),
app_model,
current_user,
str(args.last_id) if args.last_id else None,
args.limit,
)
adapter = TypeAdapter(SavedMessageItem)
items = [adapter.validate_python(message, from_attributes=True) for message in pagination.data]
@@ -61,7 +65,7 @@ class SavedMessageListApi(InstalledAppResource):
payload = SavedMessageCreatePayload.model_validate(console_ns.payload or {})
try:
SavedMessageService.save(app_model, current_user, str(payload.message_id), session=db.session())
SavedMessageService.save(db.session(), app_model, current_user, str(payload.message_id))
except MessageNotExistsError:
raise NotFound("Message Not Exists.")
@@ -84,6 +88,6 @@ class SavedMessageApi(InstalledAppResource):
if app_model.mode != "completion":
raise NotCompletionAppError()
SavedMessageService.delete(app_model, current_user, message_id_str, session=db.session())
SavedMessageService.delete(db.session(), app_model, current_user, message_id_str)
return "", 204
+5 -5
View File
@@ -431,7 +431,7 @@ class TrialAppWorkflowRunApi(TrialAppResource):
invoke_from=InvokeFrom.EXPLORE,
streaming=True,
)
RecommendedAppService.add_trial_app_record(app_id, user_id, session=session)
RecommendedAppService.add_trial_app_record(session, app_id, user_id)
# response-contract:ignore compact_generate_response
return helper.compact_generate_response(response)
except ProviderTokenNotInitError as ex:
@@ -511,7 +511,7 @@ class TrialChatApi(TrialAppResource):
invoke_from=InvokeFrom.EXPLORE,
streaming=True,
)
RecommendedAppService.add_trial_app_record(app_id, user_id, session=session)
RecommendedAppService.add_trial_app_record(session, app_id, user_id)
# response-contract:ignore compact_generate_response
return helper.compact_generate_response(response)
except services.errors.conversation.ConversationNotExistsError:
@@ -593,7 +593,7 @@ class TrialChatAudioApi(TrialAppResource):
user_id = current_user.id
response = AudioService.transcript_asr(app_model=app_model, file=file, end_user=None)
RecommendedAppService.add_trial_app_record(app_id, user_id, session=db.session())
RecommendedAppService.add_trial_app_record(db.session(), app_id, user_id)
return response
except services.errors.app_model_config.AppModelConfigBrokenError:
logger.exception("App model config broken.")
@@ -654,7 +654,7 @@ class TrialChatTextApi(TrialAppResource):
voice=voice,
message_ref=message_ref,
)
RecommendedAppService.add_trial_app_record(app_id, user_id, session=db.session())
RecommendedAppService.add_trial_app_record(db.session(), app_id, user_id)
return response
except services.errors.app_model_config.AppModelConfigBrokenError:
logger.exception("App model config broken.")
@@ -713,7 +713,7 @@ class TrialCompletionApi(TrialAppResource):
streaming=streaming,
)
RecommendedAppService.add_trial_app_record(app_id, user_id, session=session)
RecommendedAppService.add_trial_app_record(session, app_id, user_id)
# response-contract:ignore compact_generate_response
return helper.compact_generate_response(response)
except services.errors.conversation.ConversationNotExistsError:
+7 -9
View File
@@ -112,7 +112,7 @@ class APIBasedExtensionAPI(Resource):
def get(self, current_tenant_id: str):
return dump_response(
APIBasedExtensionListResponse,
APIBasedExtensionService.get_all_by_tenant_id(current_tenant_id, session=db.session()),
APIBasedExtensionService.get_all_by_tenant_id(db.session(), current_tenant_id),
)
@console_ns.doc("create_api_based_extension")
@@ -133,7 +133,7 @@ class APIBasedExtensionAPI(Resource):
api_key=payload.api_key,
)
extension = APIBasedExtensionService.save(extension_data, session=db.session())
extension = APIBasedExtensionService.save(db.session(), extension_data)
return APIBasedExtensionResponse(
id=extension.id,
name=extension.name,
@@ -158,9 +158,7 @@ class APIBasedExtensionDetailAPI(Resource):
return dump_response(
APIBasedExtensionResponse,
APIBasedExtensionService.get_with_tenant_id(
current_tenant_id, api_based_extension_id, session=db.session()
),
APIBasedExtensionService.get_with_tenant_id(db.session(), current_tenant_id, api_based_extension_id),
)
@console_ns.doc("update_api_based_extension")
@@ -176,7 +174,7 @@ class APIBasedExtensionDetailAPI(Resource):
api_based_extension_id = str(id)
extension_data_from_db = APIBasedExtensionService.get_with_tenant_id(
current_tenant_id, api_based_extension_id, session=db.session()
db.session(), current_tenant_id, api_based_extension_id
)
payload = APIBasedExtensionPayload.model_validate(console_ns.payload or {})
@@ -189,7 +187,7 @@ class APIBasedExtensionDetailAPI(Resource):
extension_data_from_db.api_key = payload.api_key
api_key_for_response = payload.api_key
APIBasedExtensionService.save(extension_data_from_db, session=db.session())
APIBasedExtensionService.save(db.session(), extension_data_from_db)
return APIBasedExtensionResponse(
id=extension_data_from_db.id,
name=extension_data_from_db.name,
@@ -210,9 +208,9 @@ class APIBasedExtensionDetailAPI(Resource):
api_based_extension_id = str(id)
extension_data_from_db = APIBasedExtensionService.get_with_tenant_id(
current_tenant_id, api_based_extension_id, session=db.session()
db.session(), current_tenant_id, api_based_extension_id
)
APIBasedExtensionService.delete(extension_data_from_db, session=db.session())
APIBasedExtensionService.delete(db.session(), extension_data_from_db)
return "", 204
+4 -4
View File
@@ -27,7 +27,6 @@ from controllers.console.wraps import (
)
from extensions.ext_database import db
from fields.file_fields import FileResponse, UploadConfig
from libs.helper import dump_response
from libs.login import login_required
from models import Account
from services.file_service import FileService
@@ -118,7 +117,8 @@ class FileApi(Resource):
except services.errors.file.BlockedFileExtensionError as blocked_extension_error:
raise BlockedFileExtensionError(blocked_extension_error.description)
return dump_response(FileResponse, upload_file), 201
response = FileResponse.model_validate(upload_file, from_attributes=True)
return response.model_dump(mode="json"), 201
@console_ns.route("/files/<uuid:file_id>/preview")
@@ -131,7 +131,7 @@ class FilePreviewApi(Resource):
def get(self, current_tenant_id: str, file_id: UUID):
file_id_str = str(file_id)
text = FileService(db.engine).get_file_preview(file_id_str, current_tenant_id)
return TextContentResponse(content=text).model_dump(mode="json")
return {"content": text}
@console_ns.route("/files/support-type")
@@ -141,4 +141,4 @@ class FileSupportTypeApi(Resource):
@account_initialization_required
@console_ns.response(200, "Success", console_ns.models[AllowedExtensionsResponse.__name__])
def get(self):
return AllowedExtensionsResponse(allowed_extensions=list(DOCUMENT_EXTENSIONS)).model_dump(mode="json")
return {"allowed_extensions": list(DOCUMENT_EXTENSIONS)}
+1 -2
View File
@@ -13,7 +13,7 @@ from services.account_service import RegisterService, TenantService
from .error import AlreadySetupError, NotInitValidateError
from .init_validate import get_init_validate_status
from .wraps import mark_setup_completed, only_edition_self_hosted
from .wraps import only_edition_self_hosted
class SetupRequestPayload(BaseModel):
@@ -96,7 +96,6 @@ def setup_system(payload: SetupRequestPayload) -> SetupResponse:
language=payload.language,
session=db.session(),
)
mark_setup_completed()
return SetupResponse(result="success")
@@ -148,8 +148,6 @@ class PublishWorkflowPayload(BaseModel):
"""Payload for publishing snippet workflow."""
knowledge_base_setting: dict[str, Any] | None = Field(default=None)
marked_name: str | None = Field(default=None, max_length=20)
marked_comment: str | None = Field(default=None, max_length=100)
class SnippetImportPayload(BaseModel):
+1 -2
View File
@@ -69,8 +69,7 @@ def handle_user_connect(sid, data):
if not workflow_id:
return {"msg": "workflow_id is required"}, 400
with sio.app.app_context():
result = collaboration_service.authorize_and_join_workflow_room(workflow_id, sid, session=db.session())
result = collaboration_service.authorize_and_join_workflow_room(workflow_id, sid, session=db.session())
if not result:
return {"msg": "unauthorized"}, 401
+4 -12
View File
@@ -1,9 +1,8 @@
import logging
from collections.abc import Mapping
from typing import Any
from flask_restx import Resource
from pydantic import Field, RootModel
from pydantic import RootModel
from controllers.common.schema import register_response_schema_models
from controllers.console.wraps import (
@@ -11,7 +10,6 @@ from controllers.console.wraps import (
setup_required,
)
from core.schemas.schema_manager import SchemaManager
from fields.base import ResponseModel
from libs.login import login_required
from . import console_ns
@@ -19,17 +17,11 @@ from . import console_ns
logger = logging.getLogger(__name__)
class SchemaDefinitionItemResponse(ResponseModel):
name: str
label: str
schema_: Mapping[str, Any] = Field(alias="schema")
class SchemaDefinitionsResponse(RootModel[Any]):
root: Any
class SchemaDefinitionsResponse(RootModel[list[SchemaDefinitionItemResponse]]):
pass
register_response_schema_models(console_ns, SchemaDefinitionItemResponse, SchemaDefinitionsResponse)
register_response_schema_models(console_ns, SchemaDefinitionsResponse)
@console_ns.route("/spec/schema-definitions")
+1 -1
View File
@@ -137,7 +137,7 @@ class TagListApi(Resource):
def get(self, current_tenant_id: str):
raw_args = request.args.to_dict()
param = TagListQueryParam.model_validate(raw_args)
tags = TagService.get_tags(param.type, current_tenant_id, param.keyword, session=db.session())
tags = TagService.get_tags(db.session(), param.type, current_tenant_id, param.keyword)
return dump_response(TagListResponse, tags), 200
+1 -1
View File
@@ -135,7 +135,7 @@ def _normalize_enum_value(value: object) -> str:
def _count_new_member_invites(tenant_id: str, emails: list[str]) -> int:
new_member_count = 0
for email in emails:
account = AccountService.get_account_by_email_with_case_fallback(email, session=db.session())
account = AccountService.get_account_by_email_with_case_fallback(db.session(), email)
if not account:
new_member_count += 1
continue
@@ -5,7 +5,7 @@ from flask import request, send_file
from flask_restx import Resource
from pydantic import BaseModel, Field, field_validator
from controllers.common.fields import SimpleResultResponse, ValidationResultResponse
from controllers.common.fields import BinaryFileResponse, SimpleResultResponse
from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models
from controllers.console import console_ns
from controllers.console.wraps import (
@@ -22,7 +22,8 @@ from extensions.ext_database import db
from fields.base import ResponseModel
from graphon.model_runtime.entities.model_entities import ModelType
from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
from libs.helper import dump_response, uuid_value
from graphon.model_runtime.utils.encoders import jsonable_encoder
from libs.helper import uuid_value
from libs.login import login_required
from models import Account
from services.billing_service import BillingService
@@ -91,8 +92,13 @@ class ModelProviderListResponse(ResponseModel):
data: list[ProviderResponse]
class ProviderCredentialsResponse(ResponseModel):
credentials: dict[str, Any] | None = None
class ProviderCredentialResponse(ResponseModel):
credentials: dict[str, Any] | None = Field(default=None)
class ProviderCredentialValidateResponse(ResponseModel):
result: Literal["success", "error"]
error: str | None = None
class ModelProviderPaymentCheckoutUrlResponse(ResponseModel):
@@ -112,20 +118,19 @@ register_schema_models(
)
register_response_schema_models(
console_ns,
BinaryFileResponse,
SimpleResultResponse,
ModelProviderListResponse,
ProviderCredentialsResponse,
ValidationResultResponse,
ModelProviderPaymentCheckoutUrlResponse,
ProviderCredentialResponse,
ProviderCredentialValidateResponse,
)
@console_ns.route("/workspaces/current/model-providers")
class ModelProviderListApi(Resource):
@console_ns.doc(params=query_params_from_model(ParserModelList))
@console_ns.response(
200, "Model providers retrieved successfully", console_ns.models[ModelProviderListResponse.__name__]
)
@console_ns.response(200, "Success", console_ns.models[ModelProviderListResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -137,17 +142,13 @@ class ModelProviderListApi(Resource):
model_provider_service = ModelProviderService()
provider_list = model_provider_service.get_provider_list(tenant_id=tenant_id, model_type=args.model_type)
return ModelProviderListResponse(data=provider_list).model_dump(mode="json")
return jsonable_encoder({"data": provider_list})
@console_ns.route("/workspaces/current/model-providers/<path:provider>/credentials")
class ModelProviderCredentialApi(Resource):
@console_ns.doc(params=query_params_from_model(ParserCredentialId))
@console_ns.response(
200,
"Provider credentials retrieved successfully",
console_ns.models[ProviderCredentialsResponse.__name__],
)
@console_ns.response(200, "Success", console_ns.models[ProviderCredentialResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -162,7 +163,7 @@ class ModelProviderCredentialApi(Resource):
tenant_id=tenant_id, provider=provider, credential_id=args.credential_id
)
return ProviderCredentialsResponse(credentials=credentials).model_dump(mode="json")
return {"credentials": credentials}
@console_ns.expect(console_ns.models[ParserCredentialCreate.__name__])
@console_ns.response(201, "Credential created successfully", console_ns.models[SimpleResultResponse.__name__])
@@ -188,7 +189,7 @@ class ModelProviderCredentialApi(Resource):
except CredentialsValidateFailedError as ex:
raise ValueError(str(ex))
return SimpleResultResponse(result="success").model_dump(mode="json"), 201
return {"result": "success"}, 201
@console_ns.expect(console_ns.models[ParserCredentialUpdate.__name__])
@console_ns.response(200, "Credential updated successfully", console_ns.models[SimpleResultResponse.__name__])
@@ -215,7 +216,7 @@ class ModelProviderCredentialApi(Resource):
except CredentialsValidateFailedError as ex:
raise ValueError(str(ex))
return SimpleResultResponse(result="success").model_dump(mode="json")
return {"result": "success"}
@console_ns.expect(console_ns.models[ParserCredentialDelete.__name__])
@console_ns.response(204, "Credential deleted successfully")
@@ -257,7 +258,7 @@ class ModelProviderCredentialSwitchApi(Resource):
provider=provider,
credential_id=args.credential_id,
)
return SimpleResultResponse(result="success").model_dump(mode="json")
return {"result": "success"}
@console_ns.route("/workspaces/current/model-providers/<path:provider>/credentials/validate")
@@ -265,8 +266,8 @@ class ModelProviderValidateApi(Resource):
@console_ns.expect(console_ns.models[ParserCredentialValidate.__name__])
@console_ns.response(
200,
"Provider credentials validated successfully",
console_ns.models[ValidationResultResponse.__name__],
"Credential validation result",
console_ns.models[ProviderCredentialValidateResponse.__name__],
)
@setup_required
@login_required
@@ -291,10 +292,12 @@ class ModelProviderValidateApi(Resource):
result = False
error = str(ex)
if not result:
return ValidationResultResponse(result="error", error=error or "Unknown error").model_dump(mode="json")
response = {"result": "success" if result else "error"}
return ValidationResultResponse(result="success").model_dump(mode="json")
if not result:
response["error"] = error or "Unknown error"
return response
@console_ns.route("/workspaces/<string:tenant_id>/model-providers/<path:provider>/<string:icon_type>/<string:lang>")
@@ -303,9 +306,8 @@ class ModelProviderIconApi(Resource):
Get model provider icon
"""
@console_ns.response(200, "Model provider icon")
@console_ns.response(200, "Success", console_ns.models[BinaryFileResponse.__name__])
def get(self, tenant_id: str, provider: str, icon_type: str, lang: str):
# response-contract:ignore binary send_file response
model_provider_service = ModelProviderService()
icon, mimetype = model_provider_service.get_model_provider_icon(
tenant_id=tenant_id,
@@ -337,16 +339,12 @@ class PreferredProviderTypeUpdateApi(Resource):
tenant_id=tenant_id, provider=provider, preferred_provider_type=args.preferred_provider_type
)
return SimpleResultResponse(result="success").model_dump(mode="json")
return {"result": "success"}
@console_ns.route("/workspaces/current/model-providers/<path:provider>/checkout-url")
class ModelProviderPaymentCheckoutUrlApi(Resource):
@console_ns.response(
200,
"Model provider checkout URL retrieved successfully",
console_ns.models[ModelProviderPaymentCheckoutUrlResponse.__name__],
)
@console_ns.response(200, "Success", console_ns.models[ModelProviderPaymentCheckoutUrlResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -355,11 +353,11 @@ class ModelProviderPaymentCheckoutUrlApi(Resource):
def get(self, current_tenant_id: str, current_user: Account, provider: str):
if provider != "anthropic":
raise ValueError(f"provider name {provider} is invalid")
BillingService.is_tenant_owner_or_admin(current_user, session=db.session())
BillingService.is_tenant_owner_or_admin(db.session(), current_user)
data = BillingService.get_model_provider_payment_link(
provider_name=provider,
tenant_id=current_tenant_id,
account_id=current_user.id,
prefilled_email=current_user.email,
)
return dump_response(ModelProviderPaymentCheckoutUrlResponse, data)
return data
+58 -75
View File
@@ -5,7 +5,7 @@ from flask import request
from flask_restx import Resource
from pydantic import BaseModel, Field, field_validator
from controllers.common.fields import SimpleResultResponse, ValidationResultResponse
from controllers.common.fields import SimpleResultResponse
from controllers.common.schema import (
query_params_from_model,
register_enum_models,
@@ -28,6 +28,7 @@ from extensions.ext_database import db
from fields.base import ResponseModel
from graphon.model_runtime.entities.model_entities import ModelType, ParameterRule
from graphon.model_runtime.errors.validate import CredentialsValidateFailedError
from graphon.model_runtime.utils.encoders import jsonable_encoder
from libs.helper import uuid_value
from libs.login import login_required
from models import Account
@@ -62,7 +63,7 @@ class ParserDeleteModels(BaseModel):
class LoadBalancingPayload(BaseModel):
configs: list[dict[str, Any]] | None = None
configs: list[dict[str, Any]] | None = Field(default=None)
enabled: bool | None = None
@@ -139,38 +140,33 @@ class DefaultModelDataResponse(ResponseModel):
data: DefaultModelResponse | None = None
class ProviderModelListResponse(ResponseModel):
class ModelWithProviderListResponse(ResponseModel):
data: list[ModelWithProviderEntityResponse]
class AvailableModelListResponse(ResponseModel):
class ProviderWithModelsDataResponse(ResponseModel):
data: list[ProviderWithModelsResponse]
class ModelLoadBalancingConfigResponse(ResponseModel):
id: str
name: str
credentials: dict[str, Any]
credential_id: str | None = None
class ModelCredentialLoadBalancingResponse(ResponseModel):
enabled: bool
in_cooldown: bool
ttl: int
class ModelLoadBalancingResponse(ResponseModel):
enabled: bool
configs: list[ModelLoadBalancingConfigResponse]
configs: list[dict[str, Any]] = Field(default_factory=list)
class ModelCredentialResponse(ResponseModel):
credentials: dict[str, Any]
credentials: dict[str, Any] = Field(default_factory=dict)
current_credential_id: str | None = None
current_credential_name: str | None = None
load_balancing: ModelLoadBalancingResponse
load_balancing: ModelCredentialLoadBalancingResponse
available_credentials: list[CredentialConfiguration]
class ModelParameterRuleListResponse(ResponseModel):
class ModelCredentialValidateResponse(ResponseModel):
result: str
error: str | None = None
class ModelParameterRulesResponse(ResponseModel):
data: list[ParameterRule]
@@ -191,12 +187,12 @@ register_schema_models(
register_response_schema_models(
console_ns,
SimpleResultResponse,
ValidationResultResponse,
DefaultModelDataResponse,
ProviderModelListResponse,
ModelWithProviderListResponse,
ProviderWithModelsDataResponse,
ModelCredentialResponse,
ModelParameterRuleListResponse,
AvailableModelListResponse,
ModelCredentialValidateResponse,
ModelParameterRulesResponse,
)
register_enum_models(console_ns, ModelType)
@@ -205,9 +201,7 @@ register_enum_models(console_ns, ModelType)
@console_ns.route("/workspaces/current/default-model")
class DefaultModelApi(Resource):
@console_ns.doc(params=query_params_from_model(ParserGetDefault))
@console_ns.response(
200, "Default model retrieved successfully", console_ns.models[DefaultModelDataResponse.__name__]
)
@console_ns.response(200, "Success", console_ns.models[DefaultModelDataResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -220,7 +214,7 @@ class DefaultModelApi(Resource):
tenant_id=tenant_id, model_type=args.model_type
)
return DefaultModelDataResponse(data=default_model_entity).model_dump(mode="json")
return jsonable_encoder({"data": default_model_entity})
@console_ns.expect(console_ns.models[ParserPostDefault.__name__])
@console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
@@ -253,14 +247,12 @@ class DefaultModelApi(Resource):
)
raise ex
return SimpleResultResponse(result="success").model_dump(mode="json")
return {"result": "success"}
@console_ns.route("/workspaces/current/model-providers/<path:provider>/models")
class ModelProviderModelApi(Resource):
@console_ns.response(
200, "Provider models retrieved successfully", console_ns.models[ProviderModelListResponse.__name__]
)
@console_ns.response(200, "Success", console_ns.models[ModelWithProviderListResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -269,10 +261,10 @@ class ModelProviderModelApi(Resource):
model_provider_service = ModelProviderService()
models = model_provider_service.get_models_by_provider(tenant_id=tenant_id, provider=provider)
return ProviderModelListResponse(data=models).model_dump(mode="json")
return jsonable_encoder({"data": models})
@console_ns.expect(console_ns.models[ParserPostModels.__name__])
@console_ns.response(200, "Model updated successfully", console_ns.models[SimpleResultResponse.__name__])
@console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
@setup_required
@login_required
@is_admin_or_owner_required
@@ -318,7 +310,7 @@ class ModelProviderModelApi(Resource):
tenant_id=tenant_id, provider=provider, model=args.model, model_type=args.model_type
)
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
return {"result": "success"}, 200
@console_ns.expect(console_ns.models[ParserDeleteModels.__name__])
@console_ns.response(204, "Model deleted successfully")
@@ -342,11 +334,7 @@ class ModelProviderModelApi(Resource):
@console_ns.route("/workspaces/current/model-providers/<path:provider>/models/credentials")
class ModelProviderModelCredentialApi(Resource):
@console_ns.doc(params=query_params_from_model(ParserGetCredentials))
@console_ns.response(
200,
"Model credentials retrieved successfully",
console_ns.models[ModelCredentialResponse.__name__],
)
@console_ns.response(200, "Success", console_ns.models[ModelCredentialResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -391,23 +379,22 @@ class ModelProviderModelCredentialApi(Resource):
model=args.model,
)
credentials: dict[str, Any] = {}
# TODO: make this throw error when type mismatches?
if current_credential and isinstance(current_credential.get("credentials"), dict):
credentials = cast(dict[str, Any], current_credential["credentials"])
return ModelCredentialResponse(
credentials=credentials,
current_credential_id=current_credential.get("current_credential_id") if current_credential else None,
current_credential_name=current_credential.get("current_credential_name") if current_credential else None,
load_balancing=ModelLoadBalancingResponse.model_validate(
{"enabled": is_load_balancing_enabled, "configs": load_balancing_configs}
),
available_credentials=available_credentials,
).model_dump(mode="json")
return jsonable_encoder(
{
"credentials": current_credential.get("credentials") if current_credential else {},
"current_credential_id": current_credential.get("current_credential_id")
if current_credential
else None,
"current_credential_name": current_credential.get("current_credential_name")
if current_credential
else None,
"load_balancing": {"enabled": is_load_balancing_enabled, "configs": load_balancing_configs},
"available_credentials": available_credentials,
}
)
@console_ns.expect(console_ns.models[ParserCreateCredential.__name__])
@console_ns.response(201, "Model credential created successfully", console_ns.models[SimpleResultResponse.__name__])
@console_ns.response(201, "Credential created successfully", console_ns.models[SimpleResultResponse.__name__])
@setup_required
@login_required
@is_admin_or_owner_required
@@ -437,10 +424,10 @@ class ModelProviderModelCredentialApi(Resource):
)
raise ValueError(str(ex))
return SimpleResultResponse(result="success").model_dump(mode="json"), 201
return {"result": "success"}, 201
@console_ns.expect(console_ns.models[ParserUpdateCredential.__name__])
@console_ns.response(200, "Model credential updated successfully", console_ns.models[SimpleResultResponse.__name__])
@console_ns.response(200, "Credential updated successfully", console_ns.models[SimpleResultResponse.__name__])
@setup_required
@login_required
@is_admin_or_owner_required
@@ -465,7 +452,7 @@ class ModelProviderModelCredentialApi(Resource):
except CredentialsValidateFailedError as ex:
raise ValueError(str(ex))
return SimpleResultResponse(result="success").model_dump(mode="json")
return {"result": "success"}
@console_ns.expect(console_ns.models[ParserDeleteCredential.__name__])
@console_ns.response(204, "Credential deleted successfully")
@@ -511,7 +498,7 @@ class ModelProviderModelCredentialSwitchApi(Resource):
model=args.model,
credential_id=args.credential_id,
)
return SimpleResultResponse(result="success").model_dump(mode="json")
return {"result": "success"}
@console_ns.route(
@@ -533,7 +520,7 @@ class ModelProviderModelEnableApi(Resource):
tenant_id=tenant_id, provider=provider, model=args.model, model_type=args.model_type
)
return SimpleResultResponse(result="success").model_dump(mode="json")
return {"result": "success"}
@console_ns.route(
@@ -555,7 +542,7 @@ class ModelProviderModelDisableApi(Resource):
tenant_id=tenant_id, provider=provider, model=args.model, model_type=args.model_type
)
return SimpleResultResponse(result="success").model_dump(mode="json")
return {"result": "success"}
class ParserValidate(BaseModel):
@@ -572,8 +559,8 @@ class ModelProviderModelValidateApi(Resource):
@console_ns.expect(console_ns.models[ParserValidate.__name__])
@console_ns.response(
200,
"Model credentials validated successfully",
console_ns.models[ValidationResultResponse.__name__],
"Credential validation result",
console_ns.models[ModelCredentialValidateResponse.__name__],
)
@setup_required
@login_required
@@ -599,20 +586,18 @@ class ModelProviderModelValidateApi(Resource):
result = False
error = str(ex)
if not result:
return ValidationResultResponse(result="error", error=error or "").model_dump(mode="json")
response = {"result": "success" if result else "error"}
return ValidationResultResponse(result="success").model_dump(mode="json")
if not result:
response["error"] = error or ""
return response
@console_ns.route("/workspaces/current/model-providers/<path:provider>/models/parameter-rules")
class ModelProviderModelParameterRuleApi(Resource):
@console_ns.doc(params=query_params_from_model(ParserParameter))
@console_ns.response(
200,
"Model parameter rules retrieved successfully",
console_ns.models[ModelParameterRuleListResponse.__name__],
)
@console_ns.response(200, "Success", console_ns.models[ModelParameterRulesResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -625,14 +610,12 @@ class ModelProviderModelParameterRuleApi(Resource):
tenant_id=tenant_id, provider=provider, model=args.model
)
return ModelParameterRuleListResponse(data=parameter_rules).model_dump(mode="json")
return jsonable_encoder({"data": parameter_rules})
@console_ns.route("/workspaces/current/models/model-types/<string:model_type>")
class ModelProviderAvailableModelApi(Resource):
@console_ns.response(
200, "Available models retrieved successfully", console_ns.models[AvailableModelListResponse.__name__]
)
@console_ns.response(200, "Success", console_ns.models[ProviderWithModelsDataResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -641,4 +624,4 @@ class ModelProviderAvailableModelApi(Resource):
model_provider_service = ModelProviderService()
models = model_provider_service.get_models_by_model_type(tenant_id=tenant_id, model_type=model_type)
return AvailableModelListResponse(data=models).model_dump(mode="json")
return jsonable_encoder({"data": models})
+17 -27
View File
@@ -31,15 +31,7 @@ from controllers.console.wraps import (
with_current_user_id,
)
from core.helper.position_helper import is_filtered
from core.plugin.entities.bundle import PluginBundleDependency
from core.plugin.entities.parameters import PluginParameterOption
from core.plugin.entities.plugin import (
PluginCategory,
PluginDeclaration,
PluginEntity,
PluginInstallationSource,
)
from core.plugin.entities.plugin_daemon import PluginDecodeResponse, PluginInstallTask, PluginInstallTaskStartResponse
from core.plugin.entities.plugin import PluginCategory, PluginInstallationSource
from core.plugin.impl.exc import PluginDaemonClientSideError
from core.plugin.plugin_service import PluginService
from core.tools.builtin_tool.providers._positions import BuiltinToolProviderSort
@@ -307,12 +299,12 @@ class PluginCategoryListResponse(ResponseModel):
has_more: bool
class PluginBundleUploadResponse(RootModel[list[PluginBundleDependency]]):
pass
class PluginDaemonOperationResponse(RootModel[Any]):
root: Any
class PluginListResponse(ResponseModel):
plugins: list[PluginEntity]
plugins: Any
total: int
@@ -342,15 +334,15 @@ class PluginInstallationsResponse(ResponseModel):
class PluginManifestResponse(ResponseModel):
manifest: PluginDeclaration
manifest: Any
class PluginTasksResponse(ResponseModel):
tasks: list[PluginInstallTask]
tasks: Any
class PluginTaskResponse(ResponseModel):
task: PluginInstallTask
task: Any
class PluginPermissionResponse(ResponseModel):
@@ -359,7 +351,7 @@ class PluginPermissionResponse(ResponseModel):
class PluginDynamicOptionsResponse(ResponseModel):
options: list[PluginParameterOption]
options: Any
class PluginOperationSuccessResponse(ResponseModel):
@@ -406,12 +398,10 @@ register_response_schema_models(
PluginCategoryBuiltinToolResponse,
PluginCategoryInstalledPluginResponse,
PluginCategoryListResponse,
PluginBundleUploadResponse,
PluginDecodeResponse,
PluginDaemonOperationResponse,
PluginDebuggingKeyResponse,
PluginDynamicOptionsResponse,
PluginInstallationsResponse,
PluginInstallTaskStartResponse,
PluginListResponse,
PluginManifestResponse,
PluginOperationSuccessResponse,
@@ -645,7 +635,7 @@ class PluginAssetApi(Resource):
@console_ns.route("/workspaces/current/plugin/upload/pkg")
class PluginUploadFromPkgApi(Resource):
@console_ns.response(200, "Success", console_ns.models[PluginDecodeResponse.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginDaemonOperationResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -666,7 +656,7 @@ class PluginUploadFromPkgApi(Resource):
@console_ns.route("/workspaces/current/plugin/upload/github")
class PluginUploadFromGithubApi(Resource):
@console_ns.expect(console_ns.models[ParserGithubUpload.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginDecodeResponse.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginDaemonOperationResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -686,7 +676,7 @@ class PluginUploadFromGithubApi(Resource):
@console_ns.route("/workspaces/current/plugin/upload/bundle")
class PluginUploadFromBundleApi(Resource):
@console_ns.response(200, "Success", console_ns.models[PluginBundleUploadResponse.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginDaemonOperationResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -707,7 +697,7 @@ class PluginUploadFromBundleApi(Resource):
@console_ns.route("/workspaces/current/plugin/install/pkg")
class PluginInstallFromPkgApi(Resource):
@console_ns.expect(console_ns.models[ParserPluginIdentifiers.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginInstallTaskStartResponse.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginDaemonOperationResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -728,7 +718,7 @@ class PluginInstallFromPkgApi(Resource):
@console_ns.route("/workspaces/current/plugin/install/github")
class PluginInstallFromGithubApi(Resource):
@console_ns.expect(console_ns.models[ParserGithubInstall.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginInstallTaskStartResponse.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginDaemonOperationResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -755,7 +745,7 @@ class PluginInstallFromGithubApi(Resource):
@console_ns.route("/workspaces/current/plugin/install/marketplace")
class PluginInstallFromMarketplaceApi(Resource):
@console_ns.expect(console_ns.models[ParserPluginIdentifiers.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginInstallTaskStartResponse.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginDaemonOperationResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -901,7 +891,7 @@ class PluginDeleteInstallTaskItemApi(Resource):
@console_ns.route("/workspaces/current/plugin/upgrade/marketplace")
class PluginUpgradeFromMarketplaceApi(Resource):
@console_ns.expect(console_ns.models[ParserMarketplaceUpgrade.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginInstallTaskStartResponse.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginDaemonOperationResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -924,7 +914,7 @@ class PluginUpgradeFromMarketplaceApi(Resource):
@console_ns.route("/workspaces/current/plugin/upgrade/github")
class PluginUpgradeFromGithubApi(Resource):
@console_ns.expect(console_ns.models[ParserGithubUpgrade.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginInstallTaskStartResponse.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginDaemonOperationResponse.__name__])
@setup_required
@login_required
@account_initialization_required
+78 -5
View File
@@ -1,9 +1,12 @@
import logging
from datetime import datetime
from typing import Any
from urllib.parse import quote
from uuid import UUID
from flask import Response, request
from flask_restx import Resource
from pydantic import Field as PydanticField
from pydantic import field_validator
from sqlalchemy.orm import Session, sessionmaker
from werkzeug.exceptions import NotFound
@@ -34,8 +37,7 @@ from controllers.console.wraps import (
from core.plugin.entities.plugin import PluginDependency
from extensions.ext_database import db
from fields.base import ResponseModel
from fields.snippet_fields import SnippetListItemResponse, SnippetPaginationResponse, SnippetResponse
from libs.helper import dump_response
from libs.helper import dump_response, to_timestamp
from libs.login import login_required
from models import Account
from models.snippet import SnippetType
@@ -63,6 +65,77 @@ class SnippetUseCountResponse(ResponseModel):
use_count: int
class SnippetTagResponse(ResponseModel):
id: str
name: str
type: str
class SnippetAccountResponse(ResponseModel):
id: str
name: str
email: str
class SnippetListItemResponse(ResponseModel):
id: str
name: str
description: str | None
type: SnippetType
version: int
use_count: int
is_published: bool
icon_info: dict[str, Any] | None
tags: list[SnippetTagResponse]
created_by: str | None
author_name: str | None
created_at: int
updated_by: str | None
updated_at: int
@field_validator("created_at", "updated_at", mode="before")
@classmethod
def _normalize_timestamp(cls, value: datetime | int | None) -> int:
timestamp = to_timestamp(value)
if timestamp is None:
raise ValueError("timestamp is required")
return timestamp
class SnippetResponse(ResponseModel):
id: str
name: str
description: str | None
type: SnippetType
version: int
use_count: int
is_published: bool
icon_info: dict[str, Any] | None
graph: dict[str, Any] = PydanticField(validation_alias="graph_dict")
input_fields: list[dict[str, Any]] = PydanticField(validation_alias="input_fields_list")
tags: list[SnippetTagResponse]
created_by: SnippetAccountResponse | None = PydanticField(validation_alias="created_by_account")
created_at: int
updated_by: SnippetAccountResponse | None = PydanticField(validation_alias="updated_by_account")
updated_at: int
@field_validator("created_at", "updated_at", mode="before")
@classmethod
def _normalize_timestamp(cls, value: datetime | int | None) -> int:
timestamp = to_timestamp(value)
if timestamp is None:
raise ValueError("timestamp is required")
return timestamp
class SnippetPaginationResponse(ResponseModel):
data: list[SnippetListItemResponse]
page: int
limit: int
total: int
has_more: bool
def _snippet_service() -> SnippetService:
return SnippetService(sessionmaker(bind=db.engine, expire_on_commit=False))
@@ -186,7 +259,7 @@ class CustomizedSnippetDetailApi(Resource):
@login_required
@account_initialization_required
@with_current_tenant_id
def get(self, current_tenant_id: str, snippet_id: UUID):
def get(self, current_tenant_id: str, snippet_id: str):
"""Get customized snippet details."""
snippet_service = _snippet_service()
snippet = snippet_service.get_snippet_by_id(
@@ -462,4 +535,4 @@ class CustomizedSnippetUseCountIncrementApi(Resource):
session.commit()
session.refresh(snippet)
return {"result": "success", "use_count": snippet.use_count}, 200
return SnippetUseCountResponse(result="success", use_count=snippet.use_count).model_dump(mode="json"), 200
File diff suppressed because it is too large Load Diff
@@ -1,19 +1,20 @@
import logging
from typing import Any
from typing import Any, Literal
from flask import make_response, redirect, request
from flask_restx import Resource
from pydantic import BaseModel, RootModel, model_validator
from pydantic import BaseModel, Field, RootModel, model_validator
from sqlalchemy.orm import sessionmaker
from werkzeug.exceptions import BadRequest, Forbidden
from configs import dify_config
from controllers.common.errors import NotFoundError
from controllers.common.fields import SimpleResultResponse
from controllers.common.fields import BinaryFileResponse, RedirectResponse, SimpleResultResponse
from controllers.common.schema import register_response_schema_models, register_schema_models
from core.entities.provider_entities import ProviderConfig
from core.entities.parameter_entities import AppSelectorScope, ModelSelectorScope, ToolSelectorScope
from core.plugin.entities.plugin_daemon import CredentialType
from core.plugin.impl.oauth import OAuthHandler
from core.tools.entities.common_entities import I18nObject
from core.trigger.entities.api_entities import (
SubscriptionBuilderApiEntity,
TriggerProviderApiEntity,
@@ -23,7 +24,7 @@ from core.trigger.entities.entities import RequestLog, SubscriptionBuilderUpdate
from core.trigger.trigger_manager import TriggerManager
from extensions.ext_database import db
from fields.base import ResponseModel
from libs.helper import dump_response
from graphon.model_runtime.utils.encoders import jsonable_encoder
from libs.login import login_required
from models.account import Account
from models.provider_ids import TriggerProviderID
@@ -58,9 +59,9 @@ class TriggerSubscriptionBuilderVerifyPayload(BaseModel):
class TriggerSubscriptionBuilderUpdatePayload(BaseModel):
name: str | None = None
parameters: dict[str, Any] | None = None
properties: dict[str, Any] | None = None
credentials: dict[str, Any] | None = None
parameters: dict[str, Any] | None = Field(default=None)
properties: dict[str, Any] | None = Field(default=None)
credentials: dict[str, Any] | None = Field(default=None)
@model_validator(mode="after")
def check_at_least_one_field(self):
@@ -70,23 +71,70 @@ class TriggerSubscriptionBuilderUpdatePayload(BaseModel):
class TriggerOAuthClientPayload(BaseModel):
client_params: dict[str, Any] | None = None
client_params: dict[str, Any] | None = Field(default=None)
enabled: bool | None = None
class TriggerOAuthAuthorizeResponse(BaseModel):
authorization_url: str
subscription_builder_id: str
subscription_builder: SubscriptionBuilderApiEntity
class TriggerProviderConfigOptionResponse(BaseModel):
value: str = Field(..., description="The value of the option")
label: I18nObject = Field(..., description="The label of the option")
class TriggerProviderConfigResponse(BaseModel):
type: Literal[
"secret-input",
"text-input",
"select",
"boolean",
"app-selector",
"model-selector",
"array[tools]",
] = Field(..., description="The type of the credentials")
name: str = Field(..., description="The name of the credentials")
scope: AppSelectorScope | ModelSelectorScope | ToolSelectorScope | None = None
required: bool = False
default: int | str | float | bool | None = None
options: list[TriggerProviderConfigOptionResponse] | None = None
multiple: bool = False
label: I18nObject | None = None
help: I18nObject | None = None
url: str | None = None
placeholder: I18nObject | None = None
class TriggerOAuthClientResponse(BaseModel):
configured: bool
system_configured: bool
custom_configured: bool
oauth_client_schema: list[TriggerProviderConfigResponse]
custom_enabled: bool
redirect_uri: str
params: dict[str, Any]
class TriggerProviderOpaqueResponse(RootModel[Any]):
root: Any
class TriggerProviderListResponse(RootModel[list[TriggerProviderApiEntity]]):
pass
root: list[TriggerProviderApiEntity]
class TriggerProviderSubscriptionListResponse(RootModel[list[TriggerProviderSubscriptionApiEntity]]):
pass
class TriggerSubscriptionListResponse(RootModel[list[TriggerProviderSubscriptionApiEntity]]):
root: list[TriggerProviderSubscriptionApiEntity]
class TriggerSubscriptionBuilderCreateResponse(ResponseModel):
subscription_builder: SubscriptionBuilderApiEntity
class TriggerVerificationResponse(ResponseModel):
class TriggerSubscriptionBuilderVerifyResponse(ResponseModel):
verified: bool
@@ -94,26 +142,6 @@ class TriggerSubscriptionBuilderLogsResponse(ResponseModel):
logs: list[RequestLog]
class TriggerOAuthAuthorizeResponse(ResponseModel):
authorization_url: str
subscription_builder_id: str
subscription_builder: SubscriptionBuilderApiEntity
class TriggerOAuthClientResponse(ResponseModel):
configured: bool
system_configured: bool
custom_configured: bool
oauth_client_schema: list[ProviderConfig]
custom_enabled: bool
redirect_uri: str
params: dict[str, Any]
class TriggerProviderErrorResponse(ResponseModel):
error: str
register_schema_models(
console_ns,
TriggerSubscriptionBuilderCreatePayload,
@@ -123,24 +151,27 @@ register_schema_models(
)
register_response_schema_models(
console_ns,
BinaryFileResponse,
RedirectResponse,
SimpleResultResponse,
TriggerOAuthAuthorizeResponse,
TriggerOAuthClientResponse,
TriggerProviderOpaqueResponse,
TriggerProviderApiEntity,
TriggerProviderErrorResponse,
TriggerProviderListResponse,
TriggerProviderSubscriptionListResponse,
TriggerSubscriptionBuilderCreateResponse,
TriggerSubscriptionBuilderLogsResponse,
TriggerProviderSubscriptionApiEntity,
TriggerSubscriptionListResponse,
SubscriptionBuilderApiEntity,
TriggerVerificationResponse,
TriggerSubscriptionBuilderCreateResponse,
TriggerSubscriptionBuilderVerifyResponse,
RequestLog,
TriggerSubscriptionBuilderLogsResponse,
)
@console_ns.route("/workspaces/current/trigger-provider/<path:provider>/icon")
class TriggerProviderIconApi(Resource):
# response-contract:ignore binary trigger provider icon
@console_ns.response(200, "Trigger provider icon")
@console_ns.response(200, "Success", console_ns.models[BinaryFileResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -151,45 +182,31 @@ class TriggerProviderIconApi(Resource):
@console_ns.route("/workspaces/current/triggers")
class TriggerProviderListApi(Resource):
@console_ns.response(
200,
"Trigger providers retrieved successfully",
console_ns.models[TriggerProviderListResponse.__name__],
)
@console_ns.response(200, "Success", console_ns.models[TriggerProviderListResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@with_current_tenant_id
def get(self, tenant_id: str):
"""List all trigger providers for the current tenant"""
return dump_response(TriggerProviderListResponse, TriggerProviderService.list_trigger_providers(tenant_id))
return jsonable_encoder(TriggerProviderService.list_trigger_providers(tenant_id))
@console_ns.route("/workspaces/current/trigger-provider/<path:provider>/info")
class TriggerProviderInfoApi(Resource):
@console_ns.response(
200,
"Trigger provider retrieved successfully",
console_ns.models[TriggerProviderApiEntity.__name__],
)
@console_ns.response(200, "Success", console_ns.models[TriggerProviderApiEntity.__name__])
@setup_required
@login_required
@account_initialization_required
@with_current_tenant_id
def get(self, tenant_id: str, provider: str):
"""Get info for a trigger provider"""
provider_entity = TriggerProviderService.get_trigger_provider(tenant_id, TriggerProviderID(provider))
return provider_entity.model_dump(mode="json")
return jsonable_encoder(TriggerProviderService.get_trigger_provider(tenant_id, TriggerProviderID(provider)))
@console_ns.route("/workspaces/current/trigger-provider/<path:provider>/subscriptions/list")
class TriggerSubscriptionListApi(Resource):
@console_ns.response(
200,
"Trigger subscriptions retrieved successfully",
console_ns.models[TriggerProviderSubscriptionListResponse.__name__],
)
@console_ns.response(404, "Trigger provider not found", console_ns.models[TriggerProviderErrorResponse.__name__])
@console_ns.response(200, "Success", console_ns.models[TriggerSubscriptionListResponse.__name__])
@setup_required
@login_required
@edit_permission_required
@@ -199,18 +216,16 @@ class TriggerSubscriptionListApi(Resource):
@with_current_tenant_id
def get(self, tenant_id: str, user: Account, provider: str):
"""List all trigger subscriptions for the current tenant's provider"""
try:
return dump_response(
TriggerProviderSubscriptionListResponse,
return jsonable_encoder(
TriggerProviderService.list_trigger_provider_subscriptions(
tenant_id=tenant_id,
provider_id=TriggerProviderID(provider),
user=user,
),
)
)
except ValueError as e:
return TriggerProviderErrorResponse(error=str(e)).model_dump(mode="json"), 404
return jsonable_encoder({"error": str(e)}), 404
except Exception as e:
logger.exception("Error listing trigger providers", exc_info=e)
raise
@@ -221,11 +236,7 @@ class TriggerSubscriptionListApi(Resource):
)
class TriggerSubscriptionBuilderCreateApi(Resource):
@console_ns.expect(console_ns.models[TriggerSubscriptionBuilderCreatePayload.__name__])
@console_ns.response(
200,
"Trigger subscription builder created successfully",
console_ns.models[TriggerSubscriptionBuilderCreateResponse.__name__],
)
@console_ns.response(200, "Success", console_ns.models[TriggerSubscriptionBuilderCreateResponse.__name__])
@setup_required
@login_required
@edit_permission_required
@@ -235,7 +246,6 @@ class TriggerSubscriptionBuilderCreateApi(Resource):
@with_current_tenant_id
def post(self, tenant_id: str, user: Account, provider: str):
"""Add a new subscription instance for a trigger provider"""
payload = TriggerSubscriptionBuilderCreatePayload.model_validate(console_ns.payload or {})
try:
@@ -246,9 +256,7 @@ class TriggerSubscriptionBuilderCreateApi(Resource):
provider_id=TriggerProviderID(provider),
credential_type=credential_type,
)
return TriggerSubscriptionBuilderCreateResponse(subscription_builder=subscription_builder).model_dump(
mode="json"
)
return jsonable_encoder({"subscription_builder": subscription_builder})
except Exception as e:
logger.exception("Error adding provider credential", exc_info=e)
raise
@@ -258,11 +266,7 @@ class TriggerSubscriptionBuilderCreateApi(Resource):
"/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/<path:subscription_builder_id>",
)
class TriggerSubscriptionBuilderGetApi(Resource):
@console_ns.response(
200,
"Trigger subscription builder retrieved successfully",
console_ns.models[SubscriptionBuilderApiEntity.__name__],
)
@console_ns.response(200, "Success", console_ns.models[SubscriptionBuilderApiEntity.__name__])
@setup_required
@login_required
@edit_permission_required
@@ -270,8 +274,9 @@ class TriggerSubscriptionBuilderGetApi(Resource):
@account_initialization_required
def get(self, provider: str, subscription_builder_id: str):
"""Get a subscription instance for a trigger provider"""
subscription_builder = TriggerSubscriptionBuilderService.get_subscription_builder_by_id(subscription_builder_id)
return subscription_builder.model_dump(mode="json")
return jsonable_encoder(
TriggerSubscriptionBuilderService.get_subscription_builder_by_id(subscription_builder_id)
)
@console_ns.route(
@@ -279,11 +284,7 @@ class TriggerSubscriptionBuilderGetApi(Resource):
)
class TriggerSubscriptionBuilderVerifyApi(Resource):
@console_ns.expect(console_ns.models[TriggerSubscriptionBuilderVerifyPayload.__name__])
@console_ns.response(
200,
"Trigger subscription builder verified successfully",
console_ns.models[TriggerVerificationResponse.__name__],
)
@console_ns.response(200, "Success", console_ns.models[TriggerSubscriptionBuilderVerifyResponse.__name__])
@setup_required
@login_required
@edit_permission_required
@@ -293,12 +294,11 @@ class TriggerSubscriptionBuilderVerifyApi(Resource):
@with_current_tenant_id
def post(self, tenant_id: str, user: Account, provider: str, subscription_builder_id: str):
"""Verify and update a subscription instance for a trigger provider"""
payload = TriggerSubscriptionBuilderVerifyPayload.model_validate(console_ns.payload or {})
try:
# Use atomic update_and_verify to prevent race conditions
result = TriggerSubscriptionBuilderService.update_and_verify_builder(
return TriggerSubscriptionBuilderService.update_and_verify_builder(
tenant_id=tenant_id,
user_id=user.id,
provider_id=TriggerProviderID(provider),
@@ -307,7 +307,6 @@ class TriggerSubscriptionBuilderVerifyApi(Resource):
credentials=payload.credentials,
),
)
return dump_response(TriggerVerificationResponse, result)
except Exception as e:
logger.exception("Error verifying provider credential", exc_info=e)
raise ValueError(str(e)) from e
@@ -318,11 +317,7 @@ class TriggerSubscriptionBuilderVerifyApi(Resource):
)
class TriggerSubscriptionBuilderUpdateApi(Resource):
@console_ns.expect(console_ns.models[TriggerSubscriptionBuilderUpdatePayload.__name__])
@console_ns.response(
200,
"Trigger subscription builder updated successfully",
console_ns.models[SubscriptionBuilderApiEntity.__name__],
)
@console_ns.response(200, "Success", console_ns.models[SubscriptionBuilderApiEntity.__name__])
@setup_required
@login_required
@edit_permission_required
@@ -331,20 +326,21 @@ class TriggerSubscriptionBuilderUpdateApi(Resource):
@with_current_tenant_id
def post(self, tenant_id: str, provider: str, subscription_builder_id: str):
"""Update a subscription instance for a trigger provider"""
payload = TriggerSubscriptionBuilderUpdatePayload.model_validate(console_ns.payload or {})
try:
return TriggerSubscriptionBuilderService.update_trigger_subscription_builder(
tenant_id=tenant_id,
provider_id=TriggerProviderID(provider),
subscription_builder_id=subscription_builder_id,
subscription_builder_updater=SubscriptionBuilderUpdater(
name=payload.name,
parameters=payload.parameters,
properties=payload.properties,
credentials=payload.credentials,
),
).model_dump(mode="json")
return jsonable_encoder(
TriggerSubscriptionBuilderService.update_trigger_subscription_builder(
tenant_id=tenant_id,
provider_id=TriggerProviderID(provider),
subscription_builder_id=subscription_builder_id,
subscription_builder_updater=SubscriptionBuilderUpdater(
name=payload.name,
parameters=payload.parameters,
properties=payload.properties,
credentials=payload.credentials,
),
)
)
except Exception as e:
logger.exception("Error updating provider credential", exc_info=e)
raise
@@ -354,11 +350,7 @@ class TriggerSubscriptionBuilderUpdateApi(Resource):
"/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/logs/<path:subscription_builder_id>",
)
class TriggerSubscriptionBuilderLogsApi(Resource):
@console_ns.response(
200,
"Trigger subscription builder logs retrieved successfully",
console_ns.models[TriggerSubscriptionBuilderLogsResponse.__name__],
)
@console_ns.response(200, "Success", console_ns.models[TriggerSubscriptionBuilderLogsResponse.__name__])
@setup_required
@login_required
@edit_permission_required
@@ -366,10 +358,9 @@ class TriggerSubscriptionBuilderLogsApi(Resource):
@account_initialization_required
def get(self, provider: str, subscription_builder_id: str):
"""Get the request logs for a subscription instance for a trigger provider"""
try:
logs = TriggerSubscriptionBuilderService.list_logs(subscription_builder_id)
return dump_response(TriggerSubscriptionBuilderLogsResponse, {"logs": logs})
return jsonable_encoder({"logs": [log.model_dump(mode="json") for log in logs]})
except Exception as e:
logger.exception("Error getting request logs for subscription builder", exc_info=e)
raise
@@ -380,9 +371,7 @@ class TriggerSubscriptionBuilderLogsApi(Resource):
)
class TriggerSubscriptionBuilderBuildApi(Resource):
@console_ns.expect(console_ns.models[TriggerSubscriptionBuilderUpdatePayload.__name__])
@console_ns.response(
200, "Trigger subscription builder built successfully", console_ns.models[SimpleResultResponse.__name__]
)
@console_ns.response(200, "Success", console_ns.models[TriggerProviderOpaqueResponse.__name__])
@setup_required
@login_required
@edit_permission_required
@@ -406,7 +395,7 @@ class TriggerSubscriptionBuilderBuildApi(Resource):
properties=payload.properties,
),
)
return SimpleResultResponse(result="success").model_dump(mode="json")
return 200
except Exception as e:
logger.exception("Error building provider credential", exc_info=e)
raise ValueError(str(e)) from e
@@ -417,9 +406,7 @@ class TriggerSubscriptionBuilderBuildApi(Resource):
)
class TriggerSubscriptionUpdateApi(Resource):
@console_ns.expect(console_ns.models[TriggerSubscriptionBuilderUpdatePayload.__name__])
@console_ns.response(
200, "Trigger subscription updated successfully", console_ns.models[SimpleResultResponse.__name__]
)
@console_ns.response(200, "Success", console_ns.models[TriggerProviderOpaqueResponse.__name__])
@setup_required
@login_required
@edit_permission_required
@@ -428,7 +415,6 @@ class TriggerSubscriptionUpdateApi(Resource):
@with_current_tenant_id
def post(self, tenant_id: str, subscription_id: str):
"""Update a subscription instance"""
request = TriggerSubscriptionBuilderUpdatePayload.model_validate(console_ns.payload or {})
subscription = TriggerProviderService.get_subscription_by_id(
@@ -454,7 +440,7 @@ class TriggerSubscriptionUpdateApi(Resource):
name=request.name,
properties=request.properties,
)
return SimpleResultResponse(result="success").model_dump(mode="json")
return 200
# For the rest cases(API_KEY, OAUTH2)
# we need to call third party provider(e.g. GitHub) to rebuild the subscription
@@ -466,7 +452,7 @@ class TriggerSubscriptionUpdateApi(Resource):
credentials=request.credentials or subscription.credentials,
parameters=request.parameters or subscription.parameters,
)
return SimpleResultResponse(result="success").model_dump(mode="json")
return 200
except ValueError as e:
raise BadRequest(str(e))
except Exception as e:
@@ -487,7 +473,6 @@ class TriggerSubscriptionDeleteApi(Resource):
@with_current_tenant_id
def post(self, tenant_id: str, subscription_id: str):
"""Delete a subscription instance"""
try:
with sessionmaker(db.engine).begin() as session:
# Delete trigger provider subscription
@@ -502,7 +487,7 @@ class TriggerSubscriptionDeleteApi(Resource):
tenant_id=tenant_id,
subscription_id=subscription_id,
)
return SimpleResultResponse(result="success").model_dump(mode="json")
return {"result": "success"}
except ValueError as e:
raise BadRequest(str(e))
except Exception as e:
@@ -512,10 +497,9 @@ class TriggerSubscriptionDeleteApi(Resource):
@console_ns.route("/workspaces/current/trigger-provider/<path:provider>/subscriptions/oauth/authorize")
class TriggerOAuthAuthorizeApi(Resource):
# response-contract:ignore cookie-bearing Flask response
@console_ns.response(
200,
"Trigger OAuth authorization URL generated successfully",
"Authorization URL retrieved successfully",
console_ns.models[TriggerOAuthAuthorizeResponse.__name__],
)
@setup_required
@@ -525,12 +509,10 @@ class TriggerOAuthAuthorizeApi(Resource):
@with_current_tenant_id
def get(self, tenant_id: str, user: Account, provider: str):
"""Initiate OAuth authorization flow for a trigger provider"""
try:
provider_id = TriggerProviderID(provider)
plugin_id = provider_id.plugin_id
provider_name = provider_id.provider_name
tenant_id = tenant_id
# Get OAuth client configuration
oauth_client_params = TriggerProviderService.get_oauth_client(
@@ -574,12 +556,15 @@ class TriggerOAuthAuthorizeApi(Resource):
system_credentials=oauth_client_params,
)
# Create response with cookie
response = make_response(
TriggerOAuthAuthorizeResponse(
authorization_url=authorization_url_response.authorization_url,
subscription_builder_id=subscription_builder.id,
subscription_builder=subscription_builder,
).model_dump(mode="json")
jsonable_encoder(
{
"authorization_url": authorization_url_response.authorization_url,
"subscription_builder_id": subscription_builder.id,
"subscription_builder": subscription_builder,
}
)
)
response.set_cookie(
"context_id",
@@ -598,8 +583,11 @@ class TriggerOAuthAuthorizeApi(Resource):
@console_ns.route("/oauth/plugin/<path:provider>/trigger/callback")
class TriggerOAuthCallbackApi(Resource):
# response-contract:ignore redirect response
@console_ns.response(302, "Redirect to OAuth callback page")
@console_ns.response(
302,
"Redirect to console OAuth callback page",
console_ns.models[RedirectResponse.__name__],
)
@setup_required
def get(self, provider: str):
"""Handle OAuth callback for trigger provider"""
@@ -665,11 +653,7 @@ class TriggerOAuthCallbackApi(Resource):
@console_ns.route("/workspaces/current/trigger-provider/<path:provider>/oauth/client")
class TriggerOAuthClientManageApi(Resource):
@console_ns.response(
200,
"Trigger OAuth client retrieved successfully",
console_ns.models[TriggerOAuthClientResponse.__name__],
)
@console_ns.response(200, "Success", console_ns.models[TriggerOAuthClientResponse.__name__])
@setup_required
@login_required
@is_admin_or_owner_required
@@ -678,7 +662,6 @@ class TriggerOAuthClientManageApi(Resource):
@with_current_tenant_id
def get(self, tenant_id: str, provider: str):
"""Get OAuth client configuration for a provider"""
try:
provider_id = TriggerProviderID(provider)
@@ -699,24 +682,24 @@ class TriggerOAuthClientManageApi(Resource):
)
provider_controller = TriggerManager.get_trigger_provider(tenant_id, provider_id)
redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/trigger/callback"
return TriggerOAuthClientResponse(
configured=bool(custom_params or system_client_exists),
system_configured=system_client_exists,
custom_configured=bool(custom_params),
oauth_client_schema=provider_controller.get_oauth_client_schema(),
custom_enabled=is_custom_enabled,
redirect_uri=redirect_uri,
params=dict(custom_params or {}),
).model_dump(mode="json")
return jsonable_encoder(
{
"configured": bool(custom_params or system_client_exists),
"system_configured": system_client_exists,
"custom_configured": bool(custom_params),
"oauth_client_schema": provider_controller.get_oauth_client_schema(),
"custom_enabled": is_custom_enabled,
"redirect_uri": redirect_uri,
"params": custom_params or {},
}
)
except Exception as e:
logger.exception("Error getting OAuth client", exc_info=e)
raise
@console_ns.expect(console_ns.models[TriggerOAuthClientPayload.__name__])
@console_ns.response(
200, "Trigger OAuth client saved successfully", console_ns.models[SimpleResultResponse.__name__]
)
@console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
@setup_required
@login_required
@is_admin_or_owner_required
@@ -725,18 +708,16 @@ class TriggerOAuthClientManageApi(Resource):
@with_current_tenant_id
def post(self, tenant_id: str, provider: str):
"""Configure custom OAuth client for a provider"""
payload = TriggerOAuthClientPayload.model_validate(console_ns.payload or {})
try:
provider_id = TriggerProviderID(provider)
result = TriggerProviderService.save_custom_oauth_client_params(
return TriggerProviderService.save_custom_oauth_client_params(
tenant_id=tenant_id,
provider_id=provider_id,
client_params=payload.client_params,
enabled=payload.enabled,
)
return dump_response(SimpleResultResponse, result)
except ValueError as e:
raise BadRequest(str(e))
@@ -744,26 +725,22 @@ class TriggerOAuthClientManageApi(Resource):
logger.exception("Error configuring OAuth client", exc_info=e)
raise
@console_ns.response(
200, "Trigger OAuth client deleted successfully", console_ns.models[SimpleResultResponse.__name__]
)
@setup_required
@login_required
@is_admin_or_owner_required
@rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.PLUGIN_PREFERENCES, resource_required=False)
@account_initialization_required
@console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
@with_current_tenant_id
def delete(self, tenant_id: str, provider: str):
"""Remove custom OAuth client configuration"""
try:
provider_id = TriggerProviderID(provider)
result = TriggerProviderService.delete_custom_oauth_client_params(
return TriggerProviderService.delete_custom_oauth_client_params(
tenant_id=tenant_id,
provider_id=provider_id,
)
return dump_response(SimpleResultResponse, result)
except ValueError as e:
raise BadRequest(str(e))
except Exception as e:
@@ -776,11 +753,7 @@ class TriggerOAuthClientManageApi(Resource):
)
class TriggerSubscriptionVerifyApi(Resource):
@console_ns.expect(console_ns.models[TriggerSubscriptionBuilderVerifyPayload.__name__])
@console_ns.response(
200,
"Trigger subscription verified successfully",
console_ns.models[TriggerVerificationResponse.__name__],
)
@console_ns.response(200, "Success", console_ns.models[TriggerSubscriptionBuilderVerifyResponse.__name__])
@setup_required
@login_required
@edit_permission_required
@@ -790,7 +763,6 @@ class TriggerSubscriptionVerifyApi(Resource):
@with_current_tenant_id
def post(self, tenant_id: str, user: Account, provider: str, subscription_id: str):
"""Verify credentials for an existing subscription (edit mode only)"""
verify_request = TriggerSubscriptionBuilderVerifyPayload.model_validate(console_ns.payload or {})
try:
@@ -801,7 +773,7 @@ class TriggerSubscriptionVerifyApi(Resource):
subscription_id=subscription_id,
credentials=verify_request.credentials,
)
return dump_response(TriggerVerificationResponse, result)
return result
except ValueError as e:
logger.warning("Credential verification failed", exc_info=e)
raise BadRequest(str(e)) from e
@@ -223,7 +223,7 @@ class TenantListApi(Resource):
def get(self, current_tenant_id: str, current_user: Account):
tenant_rows: list[tuple[Tenant, TenantAccountJoin]] = [
(tenant, membership)
for tenant, membership in TenantService.get_workspaces_for_account(current_user.id, session=db.session())
for tenant, membership in TenantService.get_workspaces_for_account(db.session(), current_user.id)
if tenant.status == TenantStatus.NORMAL
]
tenants = [tenant for tenant, _ in tenant_rows]
+3 -59
View File
@@ -4,7 +4,7 @@ import os
import time
from collections.abc import Callable
from functools import wraps
from typing import Any, Concatenate, Protocol, cast, overload
from typing import Any, Concatenate, overload
from flask import abort, request
from pydantic import BaseModel, ValidationError
@@ -46,60 +46,6 @@ ERROR_MSG_INVALID_ENCRYPTED_DATA = "Invalid encrypted data"
ERROR_MSG_INVALID_ENCRYPTED_CODE = "Invalid encrypted code"
class OnceTrueCallable[**P](Protocol):
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> bool: ...
def mark_success(self) -> None: ...
def reset_success(self) -> None: ...
def once_true[**P](func: Callable[P, bool]) -> OnceTrueCallable[P]:
"""Wrap a predicate so only a strict True result is memoized."""
has_success = False
def mark_success() -> None:
nonlocal has_success
has_success = True
def reset_success() -> None:
nonlocal has_success
has_success = False
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> bool:
nonlocal has_success
if has_success:
return True
result = func(*args, **kwargs)
if result is True:
has_success = True
return result
wrapper.mark_success = mark_success # type: ignore[attr-defined]
wrapper.reset_success = reset_success # type: ignore[attr-defined]
return cast(OnceTrueCallable[P], wrapper)
def mark_setup_completed() -> None:
"""Remember in this process that one-time self-hosted setup has completed."""
_is_setup_completed.mark_success()
@once_true
def _is_setup_completed() -> bool:
"""Check whether setup exists, caching only successful observations.
Use `once_true` instead of `@cache` because a pre-setup False result must not be memoized.
"""
return db.session.scalar(select(DifySetup).limit(1)) is not None
@overload
def account_initialization_required[T, **P, R](
view: Callable[Concatenate[T, P], R],
@@ -300,9 +246,7 @@ def setup_required[T, **P, R](
@overload
def setup_required[**P, R](view: Callable[P, R]) -> Callable[P, R]:
"""Require self-hosted bootstrap setup before serving protected routes."""
...
def setup_required[**P, R](view: Callable[P, R]) -> Callable[P, R]: ...
def setup_required[R](view: Callable[..., R]) -> Callable[..., R]:
@@ -311,7 +255,7 @@ def setup_required[R](view: Callable[..., R]) -> Callable[..., R]:
# The overloads keep Resource methods method-aware for pyrefly while
# preserving support for plain functions used in tests and utilities.
# check setup
if dify_config.EDITION == "SELF_HOSTED" and not _is_setup_completed():
if dify_config.EDITION == "SELF_HOSTED" and not db.session.scalar(select(DifySetup).limit(1)):
if os.environ.get("INIT_PASSWORD"):
raise NotInitValidateError()
raise NotSetupError()
+4 -2
View File
@@ -1,3 +1,5 @@
from mimetypes import guess_extension
from flask import request
from flask_restx import Resource
from flask_restx.api import HTTPStatus
@@ -6,7 +8,7 @@ from werkzeug.exceptions import Forbidden
import services
from core.tools.signature import verify_plugin_file_signature
from core.tools.tool_file_manager import ToolFileManager, resolve_extension
from core.tools.tool_file_manager import ToolFileManager
from core.workflow.file_reference import build_file_reference
from fields.file_fields import FileResponse
@@ -108,7 +110,7 @@ class PluginUploadFileApi(Resource):
conversation_id=args.conversation_id,
)
extension = resolve_extension(filename=tool_file.name, mimetype=tool_file.mimetype)
extension = guess_extension(tool_file.mimetype) or ".bin"
preview_url = ToolFileManager.sign_file(tool_file_id=tool_file.id, extension=extension)
# Create a dictionary with all the necessary attributes
@@ -476,7 +476,6 @@ class PluginDownloadFileRequestApi(Resource):
user_from=payload.user_from,
invoke_from=payload.invoke_from,
file_mapping=payload.file.model_dump(mode="python", exclude_none=True),
for_external=payload.for_external,
)
return BaseBackwardsInvocationResponse(
data={
+9 -12
View File
@@ -2,11 +2,11 @@ from typing import Any, Union
from flask import Response, request
from flask_restx import Resource
from pydantic import BaseModel, Field, RootModel, ValidationError
from pydantic import BaseModel, Field, ValidationError
from sqlalchemy import select
from sqlalchemy.orm import Session, sessionmaker
from controllers.common.schema import register_response_schema_models, register_schema_model
from controllers.common.schema import register_schema_model
from controllers.mcp import mcp_ns
from core.mcp import types as mcp_types
from core.mcp.server.streamable_http import handle_mcp_request, negotiate_protocol_version
@@ -33,12 +33,7 @@ class MCPRequestPayload(BaseModel):
id: int | str | None = Field(default=None, description="Request ID for tracking responses")
class MCPJSONRPCResponse(RootModel[mcp_types.JSONRPCResponse | mcp_types.JSONRPCError]):
pass
register_schema_model(mcp_ns, MCPRequestPayload)
register_response_schema_models(mcp_ns, MCPJSONRPCResponse)
@mcp_ns.route("/server/<string:server_code>/mcp")
@@ -47,10 +42,13 @@ class MCPAppApi(Resource):
@mcp_ns.doc("handle_mcp_request")
@mcp_ns.doc(description="Handle Model Context Protocol (MCP) requests for a specific server")
@mcp_ns.doc(params={"server_code": "Unique identifier for the MCP server"})
@mcp_ns.response(200, "MCP JSON-RPC response", mcp_ns.models[MCPJSONRPCResponse.__name__])
@mcp_ns.response(202, "MCP notification accepted")
@mcp_ns.response(400, "Invalid MCP request or parameters")
@mcp_ns.response(404, "Server or app not found")
@mcp_ns.doc(
responses={
200: "MCP response successfully processed",
400: "Invalid MCP request or parameters",
404: "Server or app not found",
}
)
def post(self, server_code: str):
"""Handle MCP requests for a specific server.
@@ -66,7 +64,6 @@ class MCPAppApi(Resource):
Raises:
ValidationError: Invalid request format or parameters
"""
# response-contract:ignore MCP route returns Flask Response from JSON-RPC handler
args = MCPRequestPayload.model_validate(mcp_ns.payload or {})
request_id: Union[int, str] | None = args.id
mcp_request = self._parse_mcp_request(args.model_dump(exclude_none=True))
-2
View File
@@ -2,13 +2,11 @@ from flask import Blueprint
from flask_restx import Namespace
from controllers.openapi._errors import ErrorBody, OpenApiErrorCode, OpenApiErrorFormatter
from controllers.openapi._version_gate import attach_version_gate
from libs.device_flow_security import attach_anti_framing
from libs.external_api import ExternalApi
bp = Blueprint("openapi", __name__, url_prefix="/openapi/v1")
attach_anti_framing(bp)
attach_version_gate(bp)
api = ExternalApi(
bp,
-1
View File
@@ -45,7 +45,6 @@ class OpenApiErrorCode(StrEnum):
TOO_MANY_REQUESTS = "too_many_requests"
INTERNAL_ERROR = "internal_server_error"
BAD_GATEWAY = "bad_gateway"
UPGRADE_REQUIRED = "upgrade_required"
UNKNOWN = "unknown"
# domain codes (must match the error_code attribute of the exception
# classes raised on the openapi surface)
+4 -4
View File
@@ -279,7 +279,7 @@ def _csv_string_query_schema(schema: dict[str, Any]) -> None:
class AppDescribeQuery(BaseModel):
"""`?fields=` allow-list for GET /apps/<id>.
"""`?fields=` allow-list for GET /apps/<id>/describe.
Empty / omitted all blocks. Unknown member ValidationError 422.
"""
@@ -441,7 +441,7 @@ class MemberActionResponse(BaseModel):
class TaskStopResponse(BaseModel):
"""200 body for POST /apps/<id>/tasks/<task_id>:stop. The handler always returns
"""200 body for POST /apps/<id>/tasks/<task_id>/stop. The handler always returns
{"result": "success"}, so `result` is required (no default) the generated contract
types it as a required `'success'` rather than an optional field."""
@@ -473,7 +473,7 @@ class AppDslImportPayload(BaseModel):
class AppDslExportQuery(BaseModel):
"""Query parameters for GET /apps/<app_id>/dsl."""
"""Query parameters for GET /apps/<app_id>/export."""
include_secret: bool = Field(False, description="Include encrypted secret values in the exported DSL")
workflow_id: UUIDStr | None = Field(
@@ -488,7 +488,7 @@ class AppDslExportResponse(BaseModel):
class FormSubmitResponse(BaseModel):
"""Empty 200 body for POST /apps/<id>/human-input-forms/<token>:submit. `extra='forbid'`
"""Empty 200 body for POST /apps/<id>/form/human_input/<token>. `extra='forbid'`
pins `additionalProperties: false` so the generated contract is an exact `{}` rather
than an under-annotated open object."""
-69
View File
@@ -1,69 +0,0 @@
"""Version gate: reject outdated difyctl clients on /openapi/v1 with HTTP 426.
difyctl and the ``/openapi/v1`` surface ship in lockstep. A breaking path change
(resource-oriented paths) means an outdated difyctl would call removed paths and
get a bare 404; this gate returns ``426 Upgrade Required`` with an upgrade hint
instead.
"""
from __future__ import annotations
import re
from typing import Final
from flask import Blueprint, Response, request
from packaging.version import InvalidVersion, Version
from configs import dify_config
from controllers.openapi._errors import ErrorBody, OpenApiErrorCode
_UPGRADE_HINT: Final = "Upgrade difyctl: https://docs.dify.ai/en/cli/install"
# difyctl sends `User-Agent: difyctl/<semver> (<os>; <arch>; <channel>)`.
_DIFYCTL_UA_RE = re.compile(r"^difyctl/(\d+\.\d+\.\d+(?:-[\w.]+)?)")
_PREFIX: Final = "/openapi/v1/"
# Paths a too-old client must still reach to discover that it is outdated.
_ALLOWLIST: Final = frozenset({"/openapi/v1/_version", "/openapi/v1/_health"})
def _upgrade_required_response(client_version: str, min_version: str) -> Response:
body = ErrorBody(
code=OpenApiErrorCode.UPGRADE_REQUIRED,
message=f"difyctl {client_version} is no longer supported; upgrade to >= {min_version}.",
status=426,
hint=_UPGRADE_HINT,
)
return Response(body.model_dump_json(exclude_none=True), status=426, mimetype="application/json")
def attach_version_gate(bp: Blueprint) -> None:
"""Reject difyctl clients older than ``[tool.dify] min_difyctl_version`` with 426.
Registered app-wide (``before_app_request``) rather than blueprint-scoped so it
also fires for requests to *removed* paths those no longer match an openapi
route and would 404 before a blueprint-scoped ``before_request`` ever runs. The
prefix guard scopes it back to ``/openapi/v1``. Fails open for non-difyctl or
unparseable User-Agents (only a confidently-too-old difyctl is blocked).
"""
@bp.before_app_request
def _enforce_min_client_version() -> Response | None: # pyright: ignore[reportUnusedFunction]
if not request.path.startswith(_PREFIX):
return None
if request.path in _ALLOWLIST:
return None
match = _DIFYCTL_UA_RE.match(request.headers.get("User-Agent", ""))
if match is None:
return None
try:
client_version = Version(match.group(1))
except InvalidVersion:
return None
# Compare the numeric core (major.minor.patch) only — a pre-release build
# like 0.2.0-rc.1 must not sort below the 0.2.0 floor.
min_version = dify_config.tool.dify.min_difyctl_version
if client_version.release[:3] < Version(min_version).release[:3]:
return _upgrade_required_response(match.group(1), min_version)
return None
+6 -8
View File
@@ -45,10 +45,8 @@ class AccountApi(Resource):
enforce(LIMIT_ME_PER_ACCOUNT, key=f"account:{auth_data.account_id}")
account_id_str = str(auth_data.account_id) if auth_data.account_id else None
account = AccountService.get_account_by_id(account_id_str, session=db.session()) if account_id_str else None
memberships = (
TenantService.get_account_memberships(account_id_str, session=db.session()) if account_id_str else []
)
account = AccountService.get_account_by_id(db.session(), account_id_str) if account_id_str else None
memberships = TenantService.get_account_memberships(db.session(), account_id_str) if account_id_str else []
default_ws_id = _pick_default_workspace(memberships)
return AccountResponse(
@@ -65,7 +63,7 @@ class AccountSessionsSelfApi(Resource):
@auth_router.guard(scope=Scope.FULL, allowed_token_types=frozenset({TokenType.OAUTH_ACCOUNT}))
@returns(200, RevokeResponse, description="Session revoked")
def delete(self, *, auth_data: AuthData):
revoke_oauth_token(redis_client, str(auth_data.token_id), session=db.session())
revoke_oauth_token(db.session(), redis_client, str(auth_data.token_id))
return RevokeResponse(status="revoked")
@@ -83,7 +81,7 @@ class AccountSessionsApi(Resource):
page = query.page
limit = query.limit
all_rows = list_active_sessions(ctx, now, session=db.session())
all_rows = list_active_sessions(db.session(), ctx, now)
total = len(all_rows)
sliced = all_rows[(page - 1) * limit : page * limit]
@@ -119,10 +117,10 @@ class AccountSessionByIdApi(Resource):
# 404 (not 403) on cross-subject so the endpoint doesn't leak
# token IDs that belong to other subjects.
if not token_belongs_to_subject(session_id, ctx, session=db.session()):
if not token_belongs_to_subject(db.session(), session_id, ctx):
raise NotFound("session not found")
revoke_oauth_token(redis_client, session_id, session=db.session())
revoke_oauth_token(db.session(), redis_client, session_id)
return RevokeResponse(status="revoked")
+4 -4
View File
@@ -30,7 +30,7 @@ class AppDslImportApi(Resource):
a new app.
Returns 202 when the DSL version requires an explicit confirmation step
(major version mismatch). Callers must then POST to the imports :confirm method.
(major version mismatch). Callers must then POST to the confirm endpoint.
Returns 400 when the import failed due to invalid DSL or a business error.
"""
@@ -79,7 +79,7 @@ class AppDslImportApi(Resource):
return result, 200
@openapi_ns.route("/workspaces/<string:workspace_id>/apps/imports/<string:import_id>:confirm")
@openapi_ns.route("/workspaces/<string:workspace_id>/apps/imports/<string:import_id>/confirm")
class AppDslImportConfirmApi(Resource):
"""Confirm a pending DSL import identified by ``import_id``.
@@ -119,7 +119,7 @@ class AppDslImportConfirmApi(Resource):
return result, 200
@openapi_ns.route("/apps/<string:app_id>/dsl")
@openapi_ns.route("/apps/<string:app_id>/export")
class AppDslExportApi(Resource):
"""Export an app's current draft configuration as a DSL YAML string.
@@ -154,7 +154,7 @@ class AppDslExportApi(Resource):
return AppDslExportResponse(data=data), 200
@openapi_ns.route("/apps/<string:app_id>/dependencies:check")
@openapi_ns.route("/apps/<string:app_id>/check-dependencies")
class AppDslCheckDependenciesApi(Resource):
"""Check for leaked plugin dependencies after a DSL import.
+3 -3
View File
@@ -1,4 +1,4 @@
"""POST /openapi/v1/apps/<app_id>:run — mode-agnostic runner."""
"""POST /openapi/v1/apps/<app_id>/run — mode-agnostic runner."""
from __future__ import annotations
@@ -138,7 +138,7 @@ _DISPATCH: dict[AppMode, Callable[[App, Any, AppRunRequest, Session], Any]] = {
}
@openapi_ns.route("/apps/<string:app_id>:run")
@openapi_ns.route("/apps/<string:app_id>/run")
class AppRunApi(Resource):
@auth_router.guard(
scope=Scope.APPS_RUN,
@@ -174,7 +174,7 @@ class AppRunApi(Resource):
return helper.compact_generate_response(stream_obj)
@openapi_ns.route("/apps/<string:app_id>/tasks/<string:task_id>:stop")
@openapi_ns.route("/apps/<string:app_id>/tasks/<string:task_id>/stop")
class AppRunTaskStopApi(Resource):
@auth_router.guard(
scope=Scope.APPS_RUN,
+6 -6
View File
@@ -66,13 +66,13 @@ class AppReadResource(Resource):
if is_uuid:
# ``str(parsed_uuid)`` normalises to the canonical dashed form.
app = AppService.get_visible_app_by_id(str(parsed_uuid), session=db.session())
app = AppService.get_visible_app_by_id(db.session(), str(parsed_uuid))
if app is None:
raise NotFound("app not found")
else:
if not workspace_id:
raise UnprocessableEntity("workspace_id is required for name-based lookup")
matches = AppService.find_visible_apps_by_name(name=app_id, tenant_id=workspace_id, session=db.session())
matches = AppService.find_visible_apps_by_name(db.session(), name=app_id, tenant_id=workspace_id)
if len(matches) == 0:
raise NotFound("app not found")
if len(matches) > 1:
@@ -129,7 +129,7 @@ def build_app_describe_response(app: App, fields: set[str] | None) -> AppDescrib
return AppDescribeResponse(info=info, parameters=parameters, input_schema=input_schema)
@openapi_ns.route("/apps/<string:app_id>")
@openapi_ns.route("/apps/<string:app_id>/describe")
class AppDescribeApi(AppReadResource):
@auth_router.guard(
scope=Scope.APPS_READ,
@@ -177,7 +177,7 @@ class AppListApi(Resource):
tenant_name: str | None = None
if parsed_uuid is not None:
app: App | None = AppService.get_visible_app_by_id(str(parsed_uuid), session=db.session())
app: App | None = AppService.get_visible_app_by_id(db.session(), str(parsed_uuid))
if app is None or str(app.tenant_id) != workspace_id:
return empty
if not _is_listable(app):
@@ -188,7 +188,7 @@ class AppListApi(Resource):
str(app.id), str(app.maintainer) if app.maintainer else None, str(auth_data.account_id)
):
return empty
tenant_name = TenantService.get_tenant_name(workspace_id, session=db.session())
tenant_name = TenantService.get_tenant_name(db.session(), workspace_id)
item = AppListRow(
id=str(app.id),
name=app.name,
@@ -221,7 +221,7 @@ class AppListApi(Resource):
tenant_name = None
if pagination.items:
tenant_name = TenantService.get_tenant_name(workspace_id, session=db.session())
tenant_name = TenantService.get_tenant_name(db.session(), workspace_id)
items = [
AppListRow(
@@ -55,10 +55,10 @@ class PermittedExternalAppsListApi(Resource):
return env
apps_by_id: dict[str, App] = {
str(a.id): a for a in AppService.find_visible_apps_by_ids(page_result.app_ids, session=db.session())
str(a.id): a for a in AppService.find_visible_apps_by_ids(db.session(), page_result.app_ids)
}
tenant_ids = list({str(a.tenant_id) for a in apps_by_id.values()})
tenants_by_id = {str(t.id): t for t in TenantService.get_tenants_by_ids(tenant_ids, session=db.session())}
tenants_by_id = {str(t.id): t for t in TenantService.get_tenants_by_ids(db.session(), tenant_ids)}
items: list[AppListRow] = []
for app_id in page_result.app_ids:
@@ -87,7 +87,7 @@ class PermittedExternalAppsListApi(Resource):
return env
@openapi_ns.route("/permitted-external-apps/<string:app_id>")
@openapi_ns.route("/permitted-external-apps/<string:app_id>/describe")
class PermittedExternalAppDescribeApi(Resource):
@auth_router.guard(
scope=Scope.APPS_READ_PERMITTED_EXTERNAL,
+5 -5
View File
@@ -23,7 +23,7 @@ def load_app(data: AuthData) -> None:
uuid.UUID(app_id)
except ValueError:
raise NotFound("app not found")
app = AppService.get_app_by_id(app_id, session=db.session())
app = AppService.get_app_by_id(db.session(), app_id)
if not app or app.status != AppStatus.NORMAL:
raise NotFound("app not found")
data.app = app
@@ -34,7 +34,7 @@ def load_tenant(data: AuthData) -> None:
return
if data.app is None:
raise InternalServerError("pipeline_invariant_violated: app not loaded before load_tenant")
tenant = TenantService.get_tenant_by_id(str(data.app.tenant_id), session=db.session())
tenant = TenantService.get_tenant_by_id(db.session(), str(data.app.tenant_id))
if tenant is None or tenant.status == TenantStatus.ARCHIVE:
raise Forbidden("workspace unavailable")
data.tenant = tenant
@@ -50,7 +50,7 @@ def load_tenant_from_request(data: AuthData) -> None:
uuid.UUID(workspace_id)
except ValueError:
raise NotFound("workspace not found")
tenant = TenantService.get_tenant_by_id(workspace_id, session=db.session())
tenant = TenantService.get_tenant_by_id(db.session(), workspace_id)
if tenant is None or tenant.status == TenantStatus.ARCHIVE:
raise NotFound("workspace not found")
data.tenant = tenant
@@ -59,7 +59,7 @@ def load_tenant_from_request(data: AuthData) -> None:
def load_account(data: AuthData) -> None:
if data.caller is not None:
return
account = AccountService.get_account_by_id(str(data.account_id), session=db.session())
account = AccountService.get_account_by_id(db.session(), str(data.account_id))
if account is None:
raise Unauthorized("account not found")
if data.tenant:
@@ -75,7 +75,7 @@ def load_workspace_role(data: AuthData) -> None:
return
if data.caller is not None and getattr(data.caller, "status", None) != AccountStatus.ACTIVE:
return
role = TenantService.get_account_role_in_tenant(str(data.account_id), str(data.tenant.id), session=db.session())
role = TenantService.get_account_role_in_tenant(db.session(), str(data.account_id), str(data.tenant.id))
if role is None:
return
data.tenant_role = role
+2 -2
View File
@@ -82,7 +82,7 @@ def check_app_api_enabled(data: AuthData) -> None:
def check_app_access(data: AuthData) -> None:
if data.tenant is None:
return
if not TenantService.account_belongs_to_tenant(data.account_id, data.tenant.id, session=db.session()):
if not TenantService.account_belongs_to_tenant(db.session(), data.account_id, data.tenant.id):
raise Forbidden("subject_no_app_access")
@@ -127,5 +127,5 @@ def _resolve_user_id(data: AuthData) -> str | None:
return str(data.account_id) if data.account_id is not None else None
if data.external_identity is None:
return None
account = AccountService.get_account_by_email(data.external_identity.email, session=db.session())
account = AccountService.get_account_by_email(db.session(), data.external_identity.email)
return str(account.id) if account is not None else None
+2 -2
View File
@@ -1,4 +1,4 @@
"""POST /openapi/v1/apps/<app_id>/files — upload a file for use in app inputs."""
"""POST /openapi/v1/apps/<app_id>/files/upload — upload a file for use in app inputs."""
from __future__ import annotations
@@ -26,7 +26,7 @@ from libs.oauth_bearer import Scope
from services.file_service import FileService
@openapi_ns.route("/apps/<string:app_id>/files")
@openapi_ns.route("/apps/<string:app_id>/files/upload")
class AppFileUploadApi(Resource):
@openapi_ns.doc("upload_file_for_app_input")
@openapi_ns.doc(description="Upload a file to use as an input variable when running the app")
+3 -6
View File
@@ -1,8 +1,8 @@
"""
OpenAPI bearer-authed human input form endpoints.
GET /apps/<app_id>/human-input-forms/<form_token> fetch paused form definition
POST /apps/<app_id>/human-input-forms/<form_token>:submit submit form response
GET /apps/<app_id>/form/human_input/<form_token> fetch paused form definition
POST /apps/<app_id>/form/human_input/<form_token> submit form response
"""
from __future__ import annotations
@@ -60,7 +60,7 @@ def _ensure_form_is_allowed_for_openapi(form) -> None:
raise RecipientSurfaceMismatch()
@openapi_ns.route("/apps/<string:app_id>/human-input-forms/<string:form_token>")
@openapi_ns.route("/apps/<string:app_id>/form/human_input/<string:form_token>")
class OpenApiWorkflowHumanInputFormApi(Resource):
@openapi_ns.response(200, "Form definition", openapi_ns.models[HumanInputFormDefinitionResponse.__name__])
@auth_router.guard(
@@ -79,9 +79,6 @@ class OpenApiWorkflowHumanInputFormApi(Resource):
service.ensure_form_active(form)
return _jsonify_form_definition(form)
@openapi_ns.route("/apps/<string:app_id>/human-input-forms/<string:form_token>:submit")
class OpenApiWorkflowHumanInputFormSubmitApi(Resource):
@auth_router.guard(
scope=Scope.APPS_RUN,
rbac=RBACRequirement(resource_type=RBACResourceScope.APP, scene=RBACPermission.APP_TEST_AND_RUN),
+2 -2
View File
@@ -247,6 +247,7 @@ class DeviceApproveApi(Resource):
raise BadRequest(description=str(e)) from None
ttl_days = oauth_ttl_days(tenant_id=tenant)
mint = mint_oauth_token(
db.session(),
redis_client,
subject_email=account.email,
subject_issuer=ACCOUNT_ISSUER_SENTINEL,
@@ -255,7 +256,6 @@ class DeviceApproveApi(Resource):
device_label=state.device_label,
prefix=profile.prefix,
ttl_days=ttl_days,
session=db.session(),
)
poll_payload = _build_account_poll_payload(account, tenant, mint)
@@ -342,7 +342,7 @@ def _audit_cross_ip_if_needed(state) -> None:
def _build_account_poll_payload(account, tenant, mint) -> PollPayload:
rows = TenantService.get_workspaces_for_account(str(account.id), session=db.session())
rows = TenantService.get_workspaces_for_account(db.session(), str(account.id))
workspaces = [WorkspacePayload(id=str(t.id), name=t.name, role=getattr(m, "role", "")) for t, m in rows]
# Prefer active session tenant → DB-flagged current join → first membership.
default_ws_id = None
+3 -3
View File
@@ -194,7 +194,7 @@ def _sso_complete_impl():
if state.status is not DeviceFlowStatus.PENDING:
return _device_error_redirect("sso_failed", user_code)
if AccountService.has_active_account_with_email(claims.email, session=db.session()):
if AccountService.has_active_account_with_email(db.session(), claims.email):
_emit_external_rejection_audit(
state,
_RejectedClaims(subject_email=claims.email, subject_issuer=claims.issuer),
@@ -274,7 +274,7 @@ def approve_external():
if state.status is not DeviceFlowStatus.PENDING:
raise Conflict("user_code_not_pending")
if AccountService.has_active_account_with_email(claims.subject_email, session=db.session()):
if AccountService.has_active_account_with_email(db.session(), claims.subject_email):
_emit_external_rejection_audit(state, claims, reason="email_belongs_to_dify_account")
raise Forbidden("email_belongs_to_dify_account")
@@ -293,6 +293,7 @@ def approve_external():
ttl_days = oauth_ttl_days(tenant_id=None)
mint = mint_oauth_token(
db.session(),
redis_client,
subject_email=claims.subject_email,
subject_issuer=claims.subject_issuer,
@@ -301,7 +302,6 @@ def approve_external():
device_label=state.device_label,
prefix=profile.prefix,
ttl_days=ttl_days,
session=db.session(),
)
# SSO branch of the shared PollPayload contract: account/workspace
+21 -13
View File
@@ -64,14 +64,14 @@ def _member_response(account: Account) -> MemberResponse:
def _load_tenant(workspace_id: str) -> Tenant:
tenant = TenantService.get_tenant_by_id(workspace_id, session=db.session())
tenant = TenantService.get_tenant_by_id(db.session(), workspace_id)
if tenant is None or tenant.status != TenantStatus.NORMAL:
raise NotFound("workspace not found")
return tenant
def _load_account(account_id: object) -> Account:
account = AccountService.get_account_by_id(str(account_id), session=db.session()) if account_id else None
account = AccountService.get_account_by_id(db.session(), str(account_id)) if account_id else None
if account is None:
raise RuntimeError("authenticated account_id has no Account row")
return account
@@ -94,7 +94,7 @@ class WorkspacesApi(Resource):
@auth_router.guard(scope=Scope.WORKSPACE_READ, allowed_token_types=frozenset({TokenType.OAUTH_ACCOUNT}))
@returns(200, WorkspaceListResponse, description="Workspace list")
def get(self, *, auth_data: AuthData):
rows = TenantService.get_workspaces_for_account(str(auth_data.account_id), session=db.session())
rows = TenantService.get_workspaces_for_account(db.session(), str(auth_data.account_id))
return WorkspaceListResponse(workspaces=list(starmap(_workspace_summary, rows)))
@@ -104,7 +104,7 @@ class WorkspaceByIdApi(Resource):
@auth_router.guard(scope=Scope.WORKSPACE_READ, allowed_token_types=frozenset({TokenType.OAUTH_ACCOUNT}))
@returns(200, WorkspaceDetailResponse, description="Workspace detail")
def get(self, workspace_id: str, *, auth_data: AuthData):
row = TenantService.find_workspace_for_account(str(auth_data.account_id), workspace_id, session=db.session())
row = TenantService.find_workspace_for_account(db.session(), str(auth_data.account_id), workspace_id)
# 404 (not 403) on non-member so workspace IDs don't leak across tenants.
if row is None:
raise NotFound("workspace not found")
@@ -113,7 +113,7 @@ class WorkspaceByIdApi(Resource):
return _workspace_detail(tenant, membership)
@openapi_ns.route("/workspaces/<string:workspace_id>:switch")
@openapi_ns.route("/workspaces/<string:workspace_id>/switch")
class WorkspaceSwitchApi(Resource):
"""Server-side switch — equivalent to the console's POST /workspaces/switch.
@@ -132,7 +132,7 @@ class WorkspaceSwitchApi(Resource):
except AccountNotLinkTenantError:
raise NotFound("workspace not found")
row = TenantService.find_workspace_for_account(str(auth_data.account_id), workspace_id, session=db.session())
row = TenantService.find_workspace_for_account(db.session(), str(auth_data.account_id), workspace_id)
if row is None:
raise NotFound("workspace not found")
tenant, membership = row
@@ -194,7 +194,7 @@ class WorkspaceMembersApi(Resource):
raise BadRequest(str(exc))
normalized_email = body.email.lower()
member = AccountService.get_account_by_email_with_case_fallback(normalized_email, session=db.session())
member = AccountService.get_account_by_email_with_case_fallback(db.session(), normalized_email)
if member is None:
# invite_new_member just created or fetched this account.
raise RuntimeError("invited member missing from DB after invite")
@@ -212,12 +212,11 @@ class WorkspaceMembersApi(Resource):
@openapi_ns.route("/workspaces/<string:workspace_id>/members/<string:member_id>")
class WorkspaceMemberApi(Resource):
"""Remove a member (DELETE) or change a member's role (PATCH).
"""Remove a member.
Self-removal and owner-removal are explicitly rejected by the service
layer (CannotOperateSelfError, NoPermissionError) both surface as
400 per the spec, with the service's message preserved. Owner can never be
assigned via PATCH (closed enum); admin cannot demote the standing owner.
400 per the spec, with the service's message preserved.
"""
@auth_router.guard_workspace(
@@ -229,7 +228,7 @@ class WorkspaceMemberApi(Resource):
def delete(self, workspace_id: str, member_id: str, *, auth_data: AuthData):
operator = _load_account(auth_data.account_id)
tenant = _load_tenant(workspace_id)
member = AccountService.get_account_by_id(member_id, session=db.session())
member = AccountService.get_account_by_id(db.session(), member_id)
if member is None:
raise NotFound("member not found")
@@ -244,6 +243,15 @@ class WorkspaceMemberApi(Resource):
return MemberActionResponse()
@openapi_ns.route("/workspaces/<string:workspace_id>/members/<string:member_id>/role")
class WorkspaceMemberRoleApi(Resource):
"""Change a member's role.
Owner cannot be assigned here (closed enum). Admin cannot demote the
standing owner (service NoPermissionError 400, per spec).
"""
@auth_router.guard_workspace(
scope=Scope.WORKSPACE_WRITE,
allowed_token_types=frozenset({TokenType.OAUTH_ACCOUNT}),
@@ -251,10 +259,10 @@ class WorkspaceMemberApi(Resource):
)
@returns(200, MemberActionResponse, description="Role updated")
@accepts(body=MemberRoleUpdatePayload)
def patch(self, workspace_id: str, member_id: str, *, auth_data: AuthData, body: MemberRoleUpdatePayload):
def put(self, workspace_id: str, member_id: str, *, auth_data: AuthData, body: MemberRoleUpdatePayload):
operator = _load_account(auth_data.account_id)
tenant = _load_tenant(workspace_id)
member = AccountService.get_account_by_id(member_id, session=db.session())
member = AccountService.get_account_by_id(db.session(), member_id)
if member is None:
raise NotFound("member not found")
+20 -26
View File
@@ -12,13 +12,8 @@ from controllers.service_api import service_api_ns
from controllers.service_api.wraps import validate_app_token
from extensions.ext_database import db
from extensions.ext_redis import redis_client
from fields.annotation_fields import (
Annotation,
AnnotationJobStatusDetailResponse,
AnnotationJobStatusResponse,
AnnotationList,
)
from libs.helper import dump_response
from fields.annotation_fields import Annotation, AnnotationList
from fields.base import ResponseModel
from models.model import App
from services.annotation_service import (
AppAnnotationService,
@@ -51,6 +46,12 @@ class AnnotationListQuery(BaseModel):
keyword: str = Field(default="", description="Keyword to filter annotations by question or answer content.")
class AnnotationJobStatusResponse(ResponseModel):
job_id: str
job_status: str
error_msg: str | None = None
ANNOTATION_REPLY_ACTION_PARAM = {
"description": "Action to perform: `enable` or `disable`.",
"enum": ["enable", "disable"],
@@ -66,13 +67,7 @@ register_schema_models(
Annotation,
AnnotationList,
)
register_response_schema_models(
service_api_ns,
Annotation,
AnnotationList,
AnnotationJobStatusResponse,
AnnotationJobStatusDetailResponse,
)
register_response_schema_models(service_api_ns, AnnotationJobStatusResponse)
@service_api_ns.route("/apps/annotation-reply/<string:action>")
@@ -118,7 +113,7 @@ class AnnotationReplyActionApi(Resource):
result = AppAnnotationService.enable_app_annotation(enable_args, app_model.id)
case "disable":
result = AppAnnotationService.disable_app_annotation(app_model.id)
return dump_response(AnnotationJobStatusResponse, result), 200
return result, 200
@service_api_ns.route("/apps/annotation-reply/<string:action>/status/<uuid:job_id>")
@@ -156,7 +151,7 @@ class AnnotationReplyActionStatusApi(Resource):
@service_api_ns.response(
200,
"Job status retrieved successfully",
service_api_ns.models[AnnotationJobStatusDetailResponse.__name__],
service_api_ns.models[AnnotationJobStatusResponse.__name__],
)
@validate_app_token
def get(self, app_model: App, job_id: UUID, action: str):
@@ -171,13 +166,9 @@ class AnnotationReplyActionStatusApi(Resource):
error_msg = ""
if job_status == "error":
app_annotation_error_key = f"{action}_app_annotation_error_{job_id_str}"
error_result = redis_client.get(app_annotation_error_key)
if error_result is not None:
error_msg = error_result.decode()
error_msg = redis_client.get(app_annotation_error_key).decode()
return AnnotationJobStatusDetailResponse(
job_id=job_id_str, job_status=job_status, error_msg=error_msg
).model_dump(mode="json"), 200
return {"job_id": job_id_str, "job_status": job_status, "error_msg": error_msg}, 200
@service_api_ns.route("/apps/annotations")
@@ -213,13 +204,14 @@ class AnnotationListApi(Resource):
app_model.id, query.page, query.limit, query.keyword, session=db.session()
)
annotation_models = TypeAdapter(list[Annotation]).validate_python(annotation_list, from_attributes=True)
return AnnotationList(
response = AnnotationList(
data=annotation_models,
has_more=len(annotation_list) == query.limit,
limit=query.limit,
total=total,
page=query.page,
).model_dump(mode="json")
)
return response.model_dump(mode="json")
@service_api_ns.doc(
summary="Create Annotation",
@@ -254,7 +246,8 @@ class AnnotationListApi(Resource):
annotation = AppAnnotationService.insert_app_annotation_directly(
insert_args, app_model.id, session=db.session()
)
return dump_response(Annotation, annotation), HTTPStatus.CREATED
response = Annotation.model_validate(annotation, from_attributes=True)
return response.model_dump(mode="json"), HTTPStatus.CREATED
@service_api_ns.route("/apps/annotations/<uuid:annotation_id>")
@@ -295,7 +288,8 @@ class AnnotationUpdateDeleteApi(Resource):
app_ref = AppRefService.create_app_ref(app_model)
annotation_ref = AppRefService.create_annotation_ref(app_ref, str(annotation_id))
annotation = AppAnnotationService.update_app_annotation_directly(update_args, annotation_ref, db.session())
return dump_response(Annotation, annotation)
response = Annotation.model_validate(annotation, from_attributes=True)
return response.model_dump(mode="json")
@service_api_ns.doc(
summary="Delete Annotation",
+4 -4
View File
@@ -25,7 +25,6 @@ from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate
from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
from extensions.ext_database import db
from graphon.model_runtime.errors.invoke import InvokeError
from libs.helper import dump_response
from models.model import App, EndUser
from services.app_ref_service import AppRefService
from services.audio_service import AudioService
@@ -103,7 +102,7 @@ class AudioApi(Resource):
try:
response = AudioService.transcript_asr(app_model=app_model, file=file, end_user=end_user.id)
return dump_response(AudioTranscriptResponse, response)
return response
except services.errors.app_model_config.AppModelConfigBrokenError:
logger.exception("App model config broken.")
raise AppUnavailableError()
@@ -166,7 +165,6 @@ class TextApi(Resource):
500: "Internal server error",
}
)
# TTS returns provider audio bytes, so the success response is intentionally schema-less.
@service_api_ns.response(200, "Text successfully converted to audio")
@validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON))
def post(self, app_model: App, end_user: EndUser):
@@ -188,7 +186,7 @@ class TextApi(Resource):
message_id,
end_user_id=end_user.id,
)
return AudioService.transcript_tts(
response = AudioService.transcript_tts(
app_model=app_model,
session=db.session(),
text=text,
@@ -196,6 +194,8 @@ class TextApi(Resource):
end_user=end_user.external_user_id,
message_ref=message_ref,
)
return response
except services.errors.app_model_config.AppModelConfigBrokenError:
logger.exception("App model config broken.")
raise AppUnavailableError()
+14 -8
View File
@@ -10,7 +10,7 @@ from sqlalchemy.orm import Session
from werkzeug.exceptions import BadRequest, InternalServerError, NotFound
import services
from controllers.common.fields import SimpleResultResponse
from controllers.common.fields import GeneratedAppResponse, SimpleResultResponse
from controllers.common.schema import register_response_schema_models, register_schema_models
from controllers.console.app.wraps import with_session
from controllers.service_api import service_api_ns
@@ -158,7 +158,7 @@ class ChatRequestPayload(BaseModel):
register_schema_models(service_api_ns, CompletionRequestPayload, ChatRequestPayload)
register_response_schema_models(service_api_ns, SimpleResultResponse)
register_response_schema_models(service_api_ns, GeneratedAppResponse, SimpleResultResponse)
@service_api_ns.route("/completion-messages")
@@ -201,7 +201,11 @@ class CompletionApi(Resource):
500: "Internal server error",
}
)
@service_api_ns.response(200, "Completion created successfully")
@service_api_ns.response(
200,
"Completion created successfully",
service_api_ns.models[GeneratedAppResponse.__name__],
)
@validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
@with_session
def post(self, session: Session, app_model: App, end_user: EndUser):
@@ -238,7 +242,6 @@ class CompletionApi(Resource):
streaming=streaming,
)
# response-contract:ignore compact_generate_response
return helper.compact_generate_response(response)
except services.errors.conversation.ConversationNotExistsError:
raise NotFound("Conversation Not Exists.")
@@ -301,7 +304,7 @@ class CompletionStopApi(Resource):
app_mode=AppMode.value_of(app_model.mode),
)
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
return {"result": "success"}, 200
@service_api_ns.route("/chat-messages")
@@ -351,7 +354,11 @@ class ChatApi(Resource):
500: "Internal server error",
}
)
@service_api_ns.response(200, "Message sent successfully")
@service_api_ns.response(
200,
"Message sent successfully",
service_api_ns.models[GeneratedAppResponse.__name__],
)
@validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
@with_session
def post(self, session: Session, app_model: App, end_user: EndUser):
@@ -386,7 +393,6 @@ class ChatApi(Resource):
streaming=streaming,
)
# response-contract:ignore compact_generate_response
return helper.compact_generate_response(response)
except WorkflowNotFoundError as ex:
raise NotFound(str(ex))
@@ -458,4 +464,4 @@ class ChatStopApi(Resource):
app_mode=app_mode,
)
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
return {"result": "success"}, 200
+22 -11
View File
@@ -24,7 +24,7 @@ from fields.conversation_fields import (
SimpleConversation,
)
from graphon.variables.types import SegmentType
from libs.helper import UUIDStrOrEmpty, dump_response, to_timestamp
from libs.helper import UUIDStrOrEmpty, to_timestamp
from models.model import App, AppMode, EndUser
from services.conversation_service import ConversationService
@@ -142,13 +142,15 @@ register_schema_models(
ConversationRenamePayload,
ConversationVariablesQuery,
ConversationVariableUpdatePayload,
ConversationVariableResponse,
ConversationVariableInfiniteScrollPaginationResponse,
)
register_response_schema_models(
service_api_ns,
ConversationVariableResponse,
ConversationVariableInfiniteScrollPaginationResponse,
ConversationInfiniteScrollPagination,
SimpleConversation,
ConversationVariableResponse,
ConversationVariableInfiniteScrollPaginationResponse,
)
@@ -164,9 +166,9 @@ class ConversationApi(Resource):
404: "`not_found` : Last conversation does not exist (invalid `last_id`).",
},
)
@service_api_ns.doc(params=query_params_from_model(ConversationListQuery))
@service_api_ns.doc("list_conversations")
@service_api_ns.doc(description="List all conversations for the current user")
@service_api_ns.doc(params=query_params_from_model(ConversationListQuery))
@service_api_ns.doc(
responses={
200: "Conversations retrieved successfully",
@@ -190,7 +192,7 @@ class ConversationApi(Resource):
raise NotChatAppError()
query_args = ConversationListQuery.model_validate(request.args.to_dict())
last_id = query_args.last_id or None
last_id = str(query_args.last_id) if query_args.last_id else None
try:
with sessionmaker(db.engine).begin() as session:
@@ -206,7 +208,9 @@ class ConversationApi(Resource):
adapter = TypeAdapter(SimpleConversation)
conversations = [adapter.validate_python(item, from_attributes=True) for item in pagination.data]
return ConversationInfiniteScrollPagination(
limit=pagination.limit, has_more=pagination.has_more, data=conversations
limit=pagination.limit,
has_more=pagination.has_more,
data=conversations,
).model_dump(mode="json")
except services.errors.conversation.LastConversationNotExistsError:
raise NotFound("Last Conversation Not Exists.")
@@ -297,7 +301,11 @@ class ConversationRenameApi(Resource):
conversation = ConversationService.rename(
app_model, conversation_id, end_user, payload.name, payload.auto_generate, session=db.session()
)
return dump_response(SimpleConversation, conversation)
return (
TypeAdapter(SimpleConversation)
.validate_python(conversation, from_attributes=True)
.model_dump(mode="json")
)
except services.errors.conversation.ConversationNotExistsError:
raise NotFound("Conversation Not Exists.")
@@ -314,9 +322,10 @@ class ConversationVariablesApi(Resource):
404: "`not_found` : Conversation does not exist.",
},
)
@service_api_ns.doc(params=query_params_from_model(ConversationVariablesQuery))
@service_api_ns.doc("list_conversation_variables")
@service_api_ns.doc(description="List all variables for a conversation")
@service_api_ns.doc(params={"c_id": "Conversation ID.", **query_params_from_model(ConversationVariablesQuery)})
@service_api_ns.doc(params={"c_id": "Conversation ID."})
@service_api_ns.doc(
responses={
200: "Variables retrieved successfully",
@@ -343,7 +352,7 @@ class ConversationVariablesApi(Resource):
conversation_id = str(c_id)
query_args = ConversationVariablesQuery.model_validate(request.args.to_dict())
last_id = query_args.last_id or None
last_id = str(query_args.last_id) if query_args.last_id else None
try:
pagination = ConversationService.get_conversational_variable(
@@ -355,7 +364,9 @@ class ConversationVariablesApi(Resource):
query_args.variable_name,
session=db.session(),
)
return dump_response(ConversationVariableInfiniteScrollPaginationResponse, pagination)
return ConversationVariableInfiniteScrollPaginationResponse.model_validate(
pagination, from_attributes=True
).model_dump(mode="json")
except services.errors.conversation.ConversationNotExistsError:
raise NotFound("Conversation Not Exists.")
@@ -414,7 +425,7 @@ class ConversationVariableDetailApi(Resource):
variable = ConversationService.update_conversation_variable(
app_model, conversation_id, variable_id_str, end_user, payload.value, session=db.session()
)
return dump_response(ConversationVariableResponse, variable)
return ConversationVariableResponse.model_validate(variable, from_attributes=True).model_dump(mode="json")
except services.errors.conversation.ConversationNotExistsError:
raise NotFound("Conversation Not Exists.")
except services.errors.conversation.ConversationVariableNotExistsError:
+2 -2
View File
@@ -16,7 +16,6 @@ from controllers.service_api.schema import multipart_file_params
from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
from extensions.ext_database import db
from fields.file_fields import FileResponse
from libs.helper import dump_response
from models import App, EndUser
from services.file_service import FileService
@@ -88,4 +87,5 @@ class FileApi(Resource):
except services.errors.file.UnsupportedFileTypeError:
raise UnsupportedFileTypeError()
return dump_response(FileResponse, upload_file), 201
response = FileResponse.model_validate(upload_file, from_attributes=True)
return response.model_dump(mode="json"), 201
+18 -17
View File
@@ -60,14 +60,12 @@ register_response_schema_models(
ResultResponse,
SimpleResultStringListResponse,
MessageInfiniteScrollPagination,
MessageListItem,
AppFeedbackListResponse,
)
@service_api_ns.route("/messages")
class MessageListApi(Resource):
@service_api_ns.doc("list_messages")
@service_api_ns.doc(
summary="List Conversation Messages",
description=(
@@ -78,15 +76,15 @@ class MessageListApi(Resource):
responses={
200: "Successfully retrieved conversation history.",
400: "`not_chat_app` : App mode does not match the API route.",
404: "- `not_found` : Conversation does not exist.\n- `not_found` : First message does not exist.",
404: ("- `not_found` : Conversation does not exist.\n- `not_found` : First message does not exist."),
},
)
@service_api_ns.doc(params=query_params_from_model(MessageListQuery))
@service_api_ns.doc("list_messages")
@service_api_ns.doc(description="List messages in a conversation")
@service_api_ns.doc(
responses={
200: "Messages retrieved successfully",
400: "`not_chat_app` : App mode does not match the API route.",
401: "Unauthorized - invalid API token",
404: "Conversation or first message not found",
}
@@ -107,8 +105,8 @@ class MessageListApi(Resource):
raise NotChatAppError()
query_args = MessageListQuery.model_validate(request.args.to_dict())
conversation_id = query_args.conversation_id
first_id = query_args.first_id or None
conversation_id = str(query_args.conversation_id)
first_id = str(query_args.first_id) if query_args.first_id else None
try:
pagination = MessageService.pagination_by_first_id(
@@ -117,7 +115,9 @@ class MessageListApi(Resource):
adapter = TypeAdapter(MessageListItem)
items = [adapter.validate_python(message, from_attributes=True) for message in pagination.data]
return MessageInfiniteScrollPagination(
limit=pagination.limit, has_more=pagination.has_more, data=items
limit=pagination.limit,
has_more=pagination.has_more,
data=items,
).model_dump(mode="json")
except services.errors.conversation.ConversationNotExistsError:
raise NotFound("Conversation Not Exists.")
@@ -127,20 +127,21 @@ class MessageListApi(Resource):
@service_api_ns.route("/messages/<uuid:message_id>/feedbacks")
class MessageFeedbackApi(Resource):
@expect_with_user(service_api_ns, MessageFeedbackPayload)
@service_api_ns.response(200, "Feedback submitted successfully", service_api_ns.models[ResultResponse.__name__])
@service_api_ns.doc("create_message_feedback")
@service_api_ns.doc(
summary="Submit Message Feedback",
description=(
"Submit feedback for a message. End users can rate messages as `like` or `dislike`, and "
"optionally provide text feedback. Pass `null` for `rating` to revoke previously submitted feedback."
"optionally provide text feedback. Pass `null` for `rating` to revoke previously submitted "
"feedback."
),
tags=["Feedback"],
responses={
404: "`not_found` : Message does not exist.",
},
)
@expect_with_user(service_api_ns, MessageFeedbackPayload)
@service_api_ns.response(200, "Feedback submitted successfully", service_api_ns.models[ResultResponse.__name__])
@service_api_ns.doc("create_message_feedback")
@service_api_ns.doc(description="Submit feedback for a message")
@service_api_ns.doc(params={"message_id": "Message ID."})
@service_api_ns.doc(
@@ -177,12 +178,11 @@ class MessageFeedbackApi(Resource):
@service_api_ns.route("/app/feedbacks")
class AppGetFeedbacksApi(Resource):
@service_api_ns.doc("get_app_feedbacks")
@service_api_ns.doc(
summary="List App Feedbacks",
description=(
"Retrieve a paginated list of all feedback submitted for messages in this application, including both "
"end-user and admin feedback."
"Retrieve a paginated list of all feedback submitted for messages in this application, "
"including both end-user and admin feedback."
),
tags=["Feedback"],
responses={
@@ -190,6 +190,7 @@ class AppGetFeedbacksApi(Resource):
},
)
@service_api_ns.doc(params=query_params_from_model(FeedbackListQuery))
@service_api_ns.doc("get_app_feedbacks")
@service_api_ns.doc(description="Get all feedbacks for the application")
@service_api_ns.doc(
responses={
@@ -212,12 +213,11 @@ class AppGetFeedbacksApi(Resource):
feedbacks = MessageService.get_all_messages_feedbacks(
app_model, page=query_args.page, limit=query_args.limit, session=db.session()
)
return AppFeedbackListResponse(data=feedbacks).model_dump(mode="json")
return {"data": feedbacks}
@service_api_ns.route("/messages/<uuid:message_id>/suggested")
class MessageSuggestedApi(Resource):
@service_api_ns.doc("get_suggested_questions")
@service_api_ns.doc(
summary="Get Next Suggested Questions",
description="Get next questions suggestions for the current message.",
@@ -237,6 +237,7 @@ class MessageSuggestedApi(Resource):
"Suggested questions retrieved successfully",
service_api_ns.models[SimpleResultStringListResponse.__name__],
)
@service_api_ns.doc("get_suggested_questions")
@service_api_ns.doc(description="Get suggested follow-up questions for a message")
@service_api_ns.doc(params={"message_id": "Message ID"})
@service_api_ns.doc(
@@ -275,4 +276,4 @@ class MessageSuggestedApi(Resource):
logger.exception("internal server error.")
raise InternalServerError()
return SimpleResultStringListResponse(result="success", data=questions).model_dump(mode="json")
return {"result": "success", "data": questions}
+63 -41
View File
@@ -1,24 +1,19 @@
import logging
from collections.abc import Mapping
from datetime import datetime
from typing import Literal
from typing import Literal, override
from dateutil.parser import isoparse
from flask import request
from flask_restx import Resource
from pydantic import BaseModel, Field, field_validator, model_validator
from flask_restx import Resource, fields
from pydantic import BaseModel, Field, field_validator
from pydantic.json_schema import SkipJsonSchema
from sqlalchemy.orm import Session, sessionmaker
from werkzeug.exceptions import BadRequest, InternalServerError, NotFound
from controllers.common.controller_schemas import WorkflowRunPayload as WorkflowRunPayloadBase
from controllers.common.fields import GeneratedAppResponse, SimpleResultResponse
from controllers.common.schema import (
query_params_from_model,
query_params_from_request,
register_response_schema_models,
register_schema_models,
)
from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models
from controllers.console.app.wraps import with_session
from controllers.service_api import service_api_ns
from controllers.service_api.app.error import (
@@ -47,13 +42,14 @@ from extensions.ext_database import db
from extensions.ext_redis import redis_client
from fields.base import ResponseModel
from fields.end_user_fields import SimpleEndUser
from fields.member_fields import SimpleAccountResponse
from fields.member_fields import SimpleAccount
from graphon.enums import WorkflowExecutionStatus
from graphon.graph_engine.manager import GraphEngineManager
from graphon.model_runtime.errors.invoke import InvokeError
from libs import helper
from libs.helper import dump_response, to_timestamp
from libs.helper import to_timestamp
from models.model import App, AppMode, EndUser
from models.workflow import WorkflowRun
from repositories.factory import DifyAPIRepositoryFactory
from services.app_generate_service import AppGenerateService
from services.errors.app import IsDraftWorkflowError, WorkflowIdFormatError, WorkflowNotFoundError
@@ -109,12 +105,29 @@ def _enum_value(value):
return getattr(value, "value", value)
class WorkflowRunStatusField(fields.Raw):
@override
def output(self, key, obj: WorkflowRun, **kwargs):
return _enum_value(obj.status)
class WorkflowRunOutputsField(fields.Raw):
@override
def output(self, key, obj: WorkflowRun, **kwargs):
status = _enum_value(obj.status)
if status == WorkflowExecutionStatus.PAUSED.value:
return {}
outputs = obj.outputs_dict
return outputs or {}
class WorkflowRunResponse(ResponseModel):
id: str
workflow_id: str
status: str
inputs: dict | list | str | int | float | bool | None = Field(default=None)
outputs: dict = Field(default_factory=dict, validation_alias="outputs_dict")
outputs: dict = Field(default_factory=dict)
error: str | None = None
total_steps: int | None = None
total_tokens: int | None = None
@@ -122,33 +135,11 @@ class WorkflowRunResponse(ResponseModel):
finished_at: int | None = None
elapsed_time: float | int | None = None
@field_validator("status", mode="before")
@classmethod
def _normalize_enum(cls, value):
return _enum_value(value)
@field_validator("outputs", mode="before")
@classmethod
def _normalize_outputs(cls, value):
if value is None:
return {}
if isinstance(value, dict):
return value
if isinstance(value, Mapping):
return dict(value)
return {}
@field_validator("created_at", "finished_at", mode="before")
@classmethod
def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
return to_timestamp(value)
@model_validator(mode="after")
def _clear_paused_outputs(self):
if self.status == WorkflowExecutionStatus.PAUSED.value:
self.outputs = {}
return self
class WorkflowRunForLogResponse(ResponseModel):
id: str
@@ -180,7 +171,7 @@ class WorkflowAppLogPartialResponse(ResponseModel):
details: dict | list | str | int | float | bool | None = Field(default=None)
created_from: str | None = None
created_by_role: str | None = None
created_by_account: SimpleAccountResponse | None = None
created_by_account: SimpleAccount | None = None
created_by_end_user: SimpleEndUser | None = None
created_at: int | None = None
@@ -212,6 +203,39 @@ register_response_schema_models(
)
def _serialize_workflow_run(workflow_run: WorkflowRun) -> dict:
status = _enum_value(workflow_run.status)
raw_outputs = workflow_run.outputs_dict
match raw_outputs:
case _ if status == WorkflowExecutionStatus.PAUSED.value or raw_outputs is None:
outputs: dict = {}
case dict():
outputs = raw_outputs
case _ if isinstance(raw_outputs, Mapping):
outputs = dict(raw_outputs)
case _:
outputs = {}
return WorkflowRunResponse.model_validate(
{
"id": workflow_run.id,
"workflow_id": workflow_run.workflow_id,
"status": status,
"inputs": workflow_run.inputs,
"outputs": outputs,
"error": workflow_run.error,
"total_steps": workflow_run.total_steps,
"total_tokens": workflow_run.total_tokens,
"created_at": workflow_run.created_at,
"finished_at": workflow_run.finished_at,
"elapsed_time": workflow_run.elapsed_time,
}
).model_dump(mode="json")
def _serialize_workflow_log_pagination(pagination) -> dict:
return WorkflowAppLogPaginationResponse.model_validate(pagination, from_attributes=True).model_dump(mode="json")
@service_api_ns.route("/workflows/run/<string:workflow_run_id>")
class WorkflowRunDetailApi(Resource):
@service_api_ns.doc(
@@ -264,7 +288,7 @@ class WorkflowRunDetailApi(Resource):
)
if not workflow_run:
raise NotFound("Workflow run not found.")
return dump_response(WorkflowRunResponse, workflow_run)
return _serialize_workflow_run(workflow_run)
@service_api_ns.route("/workflows/run")
@@ -349,7 +373,6 @@ class WorkflowRunApi(Resource):
streaming=streaming,
)
# response-contract:ignore compact_generate_response
return helper.compact_generate_response(response)
except ProviderTokenNotInitError as ex:
raise ProviderNotInitializeError(ex.description)
@@ -466,7 +489,6 @@ class WorkflowRunByIdApi(Resource):
streaming=streaming,
)
# response-contract:ignore compact_generate_response
return helper.compact_generate_response(response)
except WorkflowNotFoundError as ex:
raise NotFound(str(ex))
@@ -532,7 +554,7 @@ class WorkflowTaskStopApi(Resource):
# New graph engine command channel mechanism
GraphEngineManager(redis_client).send_stop_command(task_id)
return SimpleResultResponse(result="success").model_dump()
return {"result": "success"}
@service_api_ns.route("/workflows/logs")
@@ -565,7 +587,7 @@ class WorkflowAppLogApi(Resource):
Returns paginated workflow execution logs with filtering options.
"""
args = query_params_from_request(WorkflowLogQuery)
args = WorkflowLogQuery.model_validate(request.args.to_dict())
status = WorkflowExecutionStatus(args.status) if args.status else None
created_at_before = isoparse(args.created_at__before) if args.created_at__before else None
@@ -587,4 +609,4 @@ class WorkflowAppLogApi(Resource):
created_by_account=args.created_by_account,
)
return dump_response(WorkflowAppLogPaginationResponse, workflow_app_log_pagination)
return _serialize_workflow_log_pagination(workflow_app_log_pagination)
+16 -13
View File
@@ -681,10 +681,10 @@ class DatasetApi(DatasetApiResource):
dataset,
str(payload.permission) if payload.permission else None,
payload.partial_member_list,
session=db.session(),
db.session(),
)
dataset = DatasetService.update_dataset(dataset_id_str, update_data, current_user, session=session)
dataset = DatasetService.update_dataset(session, dataset_id_str, update_data, current_user)
if dataset is None:
raise NotFound("Dataset not found.")
@@ -845,7 +845,7 @@ class DocumentStatusApi(DatasetApiResource):
except ValueError as e:
raise InvalidActionError(str(e))
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
return dump_response(SimpleResultResponse, {"result": "success"}), 200
@service_api_ns.route("/datasets/tags")
@@ -876,7 +876,7 @@ class DatasetTagsApi(DatasetApiResource):
assert isinstance(current_user, Account)
cid = current_user.current_tenant_id
assert cid is not None
tags = TagService.get_tags("knowledge", cid, session=db.session())
tags = TagService.get_tags(db.session(), "knowledge", cid)
return dump_response(KnowledgeTagListResponse, tags), 200
@service_api_ns.doc(
@@ -911,8 +911,11 @@ class DatasetTagsApi(DatasetApiResource):
payload = TagCreatePayload.model_validate(service_api_ns.payload or {})
tag = TagService.save_tags(SaveTagPayload(name=payload.name, type=TagType.KNOWLEDGE), db.session())
response = KnowledgeTagResponse(id=tag.id, name=tag.name, type=tag.type, binding_count="0")
return response.model_dump(mode="json"), 200
response = dump_response(
KnowledgeTagResponse,
{"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0},
)
return response, 200
@service_api_ns.doc(
summary="Update Knowledge Tag",
@@ -950,8 +953,11 @@ class DatasetTagsApi(DatasetApiResource):
binding_count = TagService.get_tag_binding_count(tag_id, db.session(), tag_type=TagType.KNOWLEDGE)
response = KnowledgeTagResponse(id=tag.id, name=tag.name, type=tag.type, binding_count=str(binding_count))
return response.model_dump(mode="json"), 200
response = dump_response(
KnowledgeTagResponse,
{"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count},
)
return response, 200
@service_api_ns.doc(
summary="Delete Knowledge Tag",
@@ -1082,8 +1088,5 @@ class DatasetTagsBindingStatusApi(DatasetApiResource):
tags = TagService.get_tags_by_target_id(
"knowledge", current_user.current_tenant_id, str(dataset_id), db.session()
)
response = DatasetBoundTagListResponse(
data=[DatasetBoundTagResponse(id=tag.id, name=tag.name) for tag in tags],
total=len(tags),
)
return response.model_dump(mode="json"), 200
tags_list = [{"id": tag.id, "name": tag.name} for tag in tags]
return dump_response(DatasetBoundTagListResponse, {"data": tags_list, "total": len(tags)}), 200
+44 -105
View File
@@ -6,22 +6,14 @@ deprecated in generated API docs so clients migrate toward the canonical paths.
"""
import json
from collections.abc import Mapping
from contextlib import ExitStack
from copy import deepcopy
from typing import Annotated, Any, Literal, Self, override
from uuid import UUID
from flask import request, send_file
from pydantic import (
BaseModel,
Field,
GetJsonSchemaHandler,
ValidationError,
WithJsonSchema,
field_validator,
model_validator,
)
from pydantic.json_schema import SkipJsonSchema
from pydantic import BaseModel, Field, GetJsonSchemaHandler, WithJsonSchema, field_validator, model_validator
from sqlalchemy import desc, func, select
from werkzeug.exceptions import Forbidden, NotFound
@@ -34,10 +26,9 @@ from controllers.common.errors import (
TooManyFilesError,
UnsupportedFileTypeError,
)
from controllers.common.fields import UrlResponse
from controllers.common.fields import BinaryFileResponse, UrlResponse
from controllers.common.schema import (
query_params_from_model,
query_params_from_request,
register_enum_models,
register_response_schema_models,
register_schema_models,
@@ -65,7 +56,6 @@ from fields.document_fields import (
DocumentMetadataResponse,
DocumentResponse,
DocumentStatusListResponse,
normalize_enum,
)
from libs.helper import dump_response
from libs.login import current_user
@@ -291,44 +281,38 @@ class DocumentAndBatchResponse(ResponseModel):
batch: str
# Use SkipJsonSchema to support 3 metadata modes
class DocumentDetailResponse(ResponseModel):
id: str
position: int | SkipJsonSchema[None] = None
data_source_type: str | SkipJsonSchema[None] = None
data_source_info: dict[str, Any] | SkipJsonSchema[None] = None
position: int | None = None
data_source_type: str | None = None
data_source_info: dict[str, Any] | None = Field(default=None)
dataset_process_rule_id: str | None = None
dataset_process_rule: dict[str, Any] | SkipJsonSchema[None] = None
document_process_rule: dict[str, Any] | SkipJsonSchema[None] = None
name: str | SkipJsonSchema[None] = None
created_from: str | SkipJsonSchema[None] = None
created_by: str | SkipJsonSchema[None] = None
created_at: int | SkipJsonSchema[None] = None
dataset_process_rule: dict[str, Any] | None = Field(default=None)
document_process_rule: dict[str, Any] | None = Field(default=None)
name: str | None = None
created_from: str | None = None
created_by: str | None = None
created_at: int | None = None
tokens: int | None = None
indexing_status: str | SkipJsonSchema[None] = None
indexing_status: str | None = None
completed_at: int | None = None
updated_at: int | None = None
indexing_latency: float | None = None
error: str | None = None
enabled: bool | SkipJsonSchema[None] = None
enabled: bool | None = None
disabled_at: int | None = None
disabled_by: str | None = None
archived: bool | SkipJsonSchema[None] = None
archived: bool | None = None
doc_type: str | None = None
doc_metadata: list[DocumentMetadataResponse] | dict[str, Any] | None = None
segment_count: int | SkipJsonSchema[None] = None
average_segment_length: int | float | SkipJsonSchema[None] = None
hit_count: int | SkipJsonSchema[None] = None
doc_metadata: list[DocumentMetadataResponse] | None = None
segment_count: int | None = None
average_segment_length: float | None = None
hit_count: int | None = None
display_status: str | None = None
doc_form: str | SkipJsonSchema[None] = None
doc_form: str | None = None
doc_language: str | None = None
summary_index_status: str | None = None
need_summary: bool | SkipJsonSchema[None] = None
@field_validator("data_source_type", "indexing_status", "display_status", "doc_form", mode="before")
@classmethod
def _normalize_enum_fields(cls, value: Any) -> Any:
return normalize_enum(value)
need_summary: bool | None = None
register_enum_models(service_api_ns, RetrievalMethod)
@@ -348,6 +332,7 @@ register_schema_models(
)
register_response_schema_models(
service_api_ns,
BinaryFileResponse,
UrlResponse,
DocumentResponse,
DocumentAndBatchResponse,
@@ -357,13 +342,13 @@ register_response_schema_models(
)
def _create_document_by_text(tenant_id: str, dataset_id: UUID) -> tuple[Document, str]:
def _create_document_by_text(tenant_id: str, dataset_id: UUID) -> tuple[Mapping[str, object], int]:
"""Create a document from text for both canonical and legacy routes."""
payload = DocumentTextCreatePayload.model_validate(service_api_ns.payload or {})
args = payload.model_dump(exclude_none=True)
dataset_id_str = str(dataset_id)
tenant_id_str = tenant_id
tenant_id_str = str(tenant_id)
dataset = db.session.scalar(
select(Dataset).where(Dataset.tenant_id == tenant_id_str, Dataset.id == dataset_id_str).limit(1)
)
@@ -422,10 +407,10 @@ def _create_document_by_text(tenant_id: str, dataset_id: UUID) -> tuple[Document
raise ProviderNotInitializeError(ex.description)
document = documents[0]
return document, batch
return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200
def _update_document_by_text(tenant_id: str, dataset_id: UUID, document_id: UUID) -> tuple[Document, str]:
def _update_document_by_text(tenant_id: str, dataset_id: UUID, document_id: UUID) -> tuple[Mapping[str, object], int]:
"""Update a document from text for both canonical and legacy routes."""
payload = DocumentTextUpdate.model_validate(service_api_ns.payload or {})
dataset = db.session.scalar(
@@ -482,7 +467,7 @@ def _update_document_by_text(tenant_id: str, dataset_id: UUID, document_id: UUID
raise ProviderNotInitializeError(ex.description)
document = documents[0]
return document, batch
return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200
@service_api_ns.route("/datasets/<uuid:dataset_id>/document/create-by-text")
@@ -526,8 +511,7 @@ class DocumentAddByTextApi(DatasetApiResource):
@cloud_edition_billing_rate_limit_check("knowledge", "dataset")
def post(self, tenant_id: str, dataset_id: UUID):
"""Create document by text."""
document, batch = _create_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id)
return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200
return _create_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id)
@service_api_ns.route("/datasets/<uuid:dataset_id>/document/create_by_text")
@@ -559,8 +543,7 @@ class DeprecatedDocumentAddByTextApi(DatasetApiResource):
@cloud_edition_billing_rate_limit_check("knowledge", "dataset")
def post(self, tenant_id: str, dataset_id: UUID):
"""Create document by text through the deprecated underscore alias."""
document, batch = _create_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id)
return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200
return _create_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id)
@service_api_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/update-by-text")
@@ -604,8 +587,7 @@ class DocumentUpdateByTextApi(DatasetApiResource):
@cloud_edition_billing_rate_limit_check("knowledge", "dataset")
def post(self, tenant_id: str, dataset_id: UUID, document_id: UUID):
"""Update document by text."""
document, batch = _update_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200
return _update_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
@service_api_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/update_by_text")
@@ -636,8 +618,7 @@ class DeprecatedDocumentUpdateByTextApi(DatasetApiResource):
@cloud_edition_billing_rate_limit_check("knowledge", "dataset")
def post(self, tenant_id: str, dataset_id: UUID, document_id: UUID):
"""Update document by text through the deprecated underscore alias."""
document, batch = _update_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200
return _update_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
@service_api_ns.route(
@@ -786,10 +767,10 @@ class DocumentAddByFileApi(DatasetApiResource):
return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200
def _update_document_by_file(tenant_id: str, dataset_id: UUID, document_id: UUID) -> tuple[Document, str]:
def _update_document_by_file(tenant_id: str, dataset_id: UUID, document_id: UUID) -> tuple[Mapping[str, object], int]:
"""Update a document from an uploaded file for canonical and deprecated routes."""
dataset_id_str = str(dataset_id)
tenant_id_str = tenant_id
tenant_id_str = str(tenant_id)
dataset = db.session.scalar(
select(Dataset).where(Dataset.tenant_id == tenant_id_str, Dataset.id == dataset_id_str).limit(1)
)
@@ -860,7 +841,7 @@ def _update_document_by_file(tenant_id: str, dataset_id: UUID, document_id: UUID
except ProviderTokenNotInitError as ex:
raise ProviderNotInitializeError(ex.description)
document = documents[0]
return document, document.batch
return dump_response(DocumentAndBatchResponse, {"document": document, "batch": document.batch}), 200
@service_api_ns.route(
@@ -914,8 +895,7 @@ class DeprecatedDocumentUpdateByFileApi(DatasetApiResource):
@cloud_edition_billing_rate_limit_check("knowledge", "dataset")
def post(self, tenant_id: str, dataset_id: UUID, document_id: UUID):
"""Update document by file through the deprecated file-update aliases."""
document, batch = _update_document_by_file(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200
return _update_document_by_file(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
@service_api_ns.route("/datasets/<uuid:dataset_id>/documents")
@@ -948,7 +928,7 @@ class DocumentListApi(DatasetApiResource):
def get(self, tenant_id, dataset_id: UUID):
dataset_id_str = str(dataset_id)
tenant_id = str(tenant_id)
query_params = query_params_from_request(DocumentListQuery)
query_params = DocumentListQuery.model_validate(request.args.to_dict())
dataset = db.session.scalar(
select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id_str).limit(1)
)
@@ -1041,7 +1021,6 @@ class DocumentBatchDownloadZipApi(DatasetApiResource):
)
cleanup = stack.pop_all()
response.call_on_close(cleanup.close)
# response-contract:ignore binary send_file response
return response
@@ -1170,9 +1149,7 @@ class DocumentDownloadApi(DatasetApiResource):
if document.tenant_id != str(tenant_id):
raise Forbidden("No permission.")
return UrlResponse(url=DocumentService.get_document_download_url(document, db.session())).model_dump(
mode="json"
)
return {"url": DocumentService.get_document_download_url(document, db.session())}
@service_api_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")
@@ -1199,13 +1176,8 @@ class DocumentApi(DatasetApiResource):
)
@service_api_ns.doc("get_document")
@service_api_ns.doc(description="Get a specific document by ID")
@service_api_ns.doc(
params={
"dataset_id": "Knowledge base ID.",
"document_id": "Document ID.",
**query_params_from_model(DocumentGetQuery),
}
)
@service_api_ns.doc(params={"dataset_id": "Knowledge base ID.", "document_id": "Document ID."})
@service_api_ns.doc(params=query_params_from_model(DocumentGetQuery))
@service_api_ns.doc(
responses={
200: "Document retrieved successfully",
@@ -1233,14 +1205,9 @@ class DocumentApi(DatasetApiResource):
if document.tenant_id != str(tenant_id):
raise Forbidden("No permission.")
try:
query_params = query_params_from_request(DocumentGetQuery)
except ValidationError as exc:
metadata = request.args.get("metadata", "all")
raise InvalidMetadataError(f"Invalid metadata value: {metadata}") from exc
metadata = query_params.metadata
response_include: set[str] | None = None
response_exclude: set[str] | None = None
metadata = request.args.get("metadata", "all")
if metadata not in self.METADATA_CHOICES:
raise InvalidMetadataError(f"Invalid metadata value: {metadata}")
# Calculate summary_index_status if needed
summary_index_status = None
@@ -1254,10 +1221,8 @@ class DocumentApi(DatasetApiResource):
)
if metadata == "only":
response_include = {"id", "doc_type", "doc_metadata"}
response = {"id": document.id, "doc_type": document.doc_type, "doc_metadata": document.doc_metadata_details}
elif metadata == "without":
response_exclude = {"doc_type", "doc_metadata"}
dataset_process_rules = DatasetService.get_process_rules(dataset_id_str, db.session())
document_process_rules = document.dataset_process_rule.to_dict() if document.dataset_process_rule else {}
data_source_info = document.data_source_detail_dict
@@ -1330,33 +1295,8 @@ class DocumentApi(DatasetApiResource):
"need_summary": document.need_summary if document.need_summary is not None else False,
}
return DocumentDetailResponse.model_validate(response).model_dump(
mode="json",
include=response_include,
exclude=response_exclude,
)
return response
@service_api_ns.doc(
summary="Update Document by File",
description=(
"Update an existing document by uploading a new file. Re-triggers indexing — use the returned "
"`batch` ID with [Get Document Indexing Status](/api-reference/documents/"
"get-document-indexing-status) to track progress."
),
tags=["Documents"],
responses={
200: "Document updated successfully.",
400: (
"- `too_many_files` : Only one file is allowed.\n"
"- `filename_not_exists_error` : The specified filename does not exist.\n"
"- `provider_not_initialize` : No valid model provider credentials found. Please go to "
"Settings -> Model Provider to complete your provider credentials.\n"
"- `invalid_param` : Knowledge base does not exist, external datasets not supported, "
"file too large, unsupported file type, or invalid doc_form (must be `text_model`, "
"`hierarchical_model`, or `qa_model`)."
),
},
)
@service_api_ns.doc("update_document_by_file")
@service_api_ns.doc(description="Update an existing document by uploading a file")
@service_api_ns.doc(consumes=["multipart/form-data"], params=DOCUMENT_UPDATE_BY_FILE_PARAMS)
@@ -1374,8 +1314,7 @@ class DocumentApi(DatasetApiResource):
@cloud_edition_billing_rate_limit_check("knowledge", "dataset")
def patch(self, tenant_id: str, dataset_id: UUID, document_id: UUID):
"""Update document by file on the canonical document resource."""
document, batch = _update_document_by_file(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200
return _update_document_by_file(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
@service_api_ns.doc(
summary="Delete Document",
@@ -86,7 +86,7 @@ class DatasetMetadataCreateServiceApi(DatasetApiResource):
raise NotFound("Dataset not found.")
DatasetService.check_dataset_permission(dataset, current_user, db.session())
metadata = MetadataService.create_metadata(dataset_id_str, metadata_args, session=db.session())
metadata = MetadataService.create_metadata(db.session(), dataset_id_str, metadata_args)
return dump_response(DatasetMetadataResponse, metadata), 201
@service_api_ns.doc(
@@ -119,7 +119,7 @@ class DatasetMetadataCreateServiceApi(DatasetApiResource):
dataset = DatasetService.get_dataset(dataset_id_str, db.session())
if dataset is None:
raise NotFound("Dataset not found.")
metadata = MetadataService.get_dataset_metadatas(dataset, session=db.session())
metadata = MetadataService.get_dataset_metadatas(db.session(), dataset)
return dump_response(DatasetMetadataListResponse, metadata), 200
@@ -159,9 +159,7 @@ class DatasetMetadataServiceApi(DatasetApiResource):
raise NotFound("Dataset not found.")
DatasetService.check_dataset_permission(dataset, current_user, db.session())
metadata = MetadataService.update_metadata_name(
dataset_id_str, metadata_id_str, payload.name, session=db.session()
)
metadata = MetadataService.update_metadata_name(db.session(), dataset_id_str, metadata_id_str, payload.name)
return dump_response(DatasetMetadataResponse, metadata), 200
@service_api_ns.doc(
@@ -196,7 +194,7 @@ class DatasetMetadataServiceApi(DatasetApiResource):
raise NotFound("Dataset not found.")
DatasetService.check_dataset_permission(dataset, current_user, db.session())
MetadataService.delete_metadata(dataset_id_str, metadata_id_str, session=db.session())
MetadataService.delete_metadata(db.session(), dataset_id_str, metadata_id_str)
return "", 204
@@ -266,9 +264,9 @@ class DatasetMetadataBuiltInFieldActionServiceApi(DatasetApiResource):
match action:
case "enable":
MetadataService.enable_built_in_field(dataset, session=db.session())
MetadataService.enable_built_in_field(db.session(), dataset)
case "disable":
MetadataService.disable_built_in_field(dataset, session=db.session())
MetadataService.disable_built_in_field(db.session(), dataset)
return dump_response(DatasetMetadataActionResponse, {"result": "success"}), 200
@@ -312,6 +310,6 @@ class DocumentMetadataEditServiceApi(DatasetApiResource):
metadata_args = MetadataOperationData.model_validate(service_api_ns.payload or {})
MetadataService.update_documents_metadata(dataset, metadata_args, session=db.session())
MetadataService.update_documents_metadata(db.session(), dataset, metadata_args)
return dump_response(DatasetMetadataActionResponse, {"result": "success"}), 200
@@ -272,7 +272,7 @@ class PipelineRunApi(DatasetApiResource):
dataset_id_str = str(dataset_id)
# Verify dataset ownership
stmt = select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id_str)
dataset = session.scalar(stmt)
dataset = db.session.scalar(stmt)
if not dataset:
raise NotFound("Dataset not found.")
@@ -281,7 +281,7 @@ class PipelineRunApi(DatasetApiResource):
if not isinstance(current_user, Account):
raise Forbidden()
rag_pipeline_service = RagPipelineService(session)
rag_pipeline_service = RagPipelineService(db.session())
pipeline = rag_pipeline_service.get_pipeline(tenant_id=tenant_id, dataset_id=dataset_id_str)
try:
response: dict[Any, Any] | Generator[str, Any, None] = PipelineGenerateService.generate(
+2 -4
View File
@@ -180,9 +180,7 @@ class AppWebAuthPermission(Resource):
if not app_id or not app_code:
raise ValueError("appId must be provided")
require_permission_check = WebAppAuthService.is_app_require_permission_check(
app_id=app_id, session=db.session()
)
require_permission_check = WebAppAuthService.is_app_require_permission_check(db.session(), app_id=app_id)
if not require_permission_check:
return {"result": True}
@@ -203,6 +201,6 @@ class AppWebAuthPermission(Resource):
return {"result": True}
res = True
if WebAppAuthService.is_app_require_permission_check(app_id=app_id, session=db.session()):
if WebAppAuthService.is_app_require_permission_check(db.session(), app_id=app_id):
res = EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp(str(user_id), app_id)
return {"result": res}
+15 -12
View File
@@ -1,11 +1,13 @@
import logging
from flask import request
from flask_restx import fields, marshal_with
from pydantic import field_validator
from werkzeug.exceptions import InternalServerError
import services
from controllers.common.controller_schemas import TextToAudioPayload as TextToAudioPayloadBase
from controllers.common.fields import AudioBinaryResponse, AudioTranscriptResponse
from controllers.web import web_ns
from controllers.web.error import (
AppUnavailableError,
@@ -21,9 +23,8 @@ from controllers.web.error import (
from controllers.web.wraps import WebApiResource
from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
from extensions.ext_database import db
from fields.base import ResponseModel
from graphon.model_runtime.errors.invoke import InvokeError
from libs.helper import dump_response, uuid_value
from libs.helper import uuid_value
from models.model import App, EndUser
from services.app_ref_service import AppRefService
from services.audio_service import AudioService
@@ -37,10 +38,6 @@ from services.errors.audio import (
from ..common.schema import register_response_schema_models, register_schema_models
class AudioToTextResponse(ResponseModel):
text: str
class TextToAudioPayload(TextToAudioPayloadBase):
@field_validator("message_id")
@classmethod
@@ -51,13 +48,18 @@ class TextToAudioPayload(TextToAudioPayloadBase):
register_schema_models(web_ns, TextToAudioPayload)
register_response_schema_models(web_ns, AudioToTextResponse)
register_response_schema_models(web_ns, AudioBinaryResponse, AudioTranscriptResponse)
logger = logging.getLogger(__name__)
@web_ns.route("/audio-to-text")
class AudioApi(WebApiResource):
audio_to_text_response_fields = {
"text": fields.String,
}
@marshal_with(audio_to_text_response_fields)
@web_ns.doc("Audio to Text")
@web_ns.doc(description="Convert audio file to text using speech-to-text service.")
@web_ns.doc(
@@ -71,7 +73,7 @@ class AudioApi(WebApiResource):
500: "Internal Server Error",
}
)
@web_ns.response(200, "Success", web_ns.models[AudioToTextResponse.__name__])
@web_ns.response(200, "Success", web_ns.models[AudioTranscriptResponse.__name__])
def post(self, app_model: App, end_user: EndUser):
"""Convert audio to text"""
file = request.files["file"]
@@ -79,7 +81,7 @@ class AudioApi(WebApiResource):
try:
response = AudioService.transcript_asr(app_model=app_model, file=file, end_user=end_user.external_user_id)
return dump_response(AudioToTextResponse, response)
return response
except services.errors.app_model_config.AppModelConfigBrokenError:
logger.exception("App model config broken.")
raise AppUnavailableError()
@@ -120,8 +122,7 @@ class TextApi(WebApiResource):
500: "Internal Server Error",
}
)
# response-contract:ignore provider audio bytes; TODO: model binary audio response if shape is standardized.
@web_ns.response(200, "Success")
@web_ns.response(200, "Success", web_ns.models[AudioBinaryResponse.__name__])
def post(self, app_model: App, end_user: EndUser):
"""Convert text to audio"""
try:
@@ -138,7 +139,7 @@ class TextApi(WebApiResource):
message_id,
end_user_id=end_user.id,
)
return AudioService.transcript_tts(
response = AudioService.transcript_tts(
app_model=app_model,
session=db.session(),
text=text,
@@ -146,6 +147,8 @@ class TextApi(WebApiResource):
end_user=end_user.external_user_id,
message_ref=message_ref,
)
return response
except services.errors.app_model_config.AppModelConfigBrokenError:
logger.exception("App model config broken.")
raise AppUnavailableError()
+2 -4
View File
@@ -133,7 +133,6 @@ class CompletionApi(WebApiResource):
streaming=streaming,
)
# response-contract:ignore compact_generate_response
return helper.compact_generate_response(response)
except services.errors.conversation.ConversationNotExistsError:
raise NotFound("Conversation Not Exists.")
@@ -186,7 +185,7 @@ class CompletionStopApi(WebApiResource):
app_mode=AppMode.value_of(app_model.mode),
)
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
return {"result": "success"}, 200
@web_ns.route("/chat-messages")
@@ -236,7 +235,6 @@ class ChatApi(WebApiResource):
streaming=streaming,
)
# response-contract:ignore compact_generate_response
return helper.compact_generate_response(response)
except services.errors.conversation.ConversationNotExistsError:
raise NotFound("Conversation Not Exists.")
@@ -292,4 +290,4 @@ class ChatStopApi(WebApiResource):
app_mode=app_mode,
)
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
return {"result": "success"}, 200
+2 -2
View File
@@ -13,7 +13,6 @@ from controllers.web import web_ns
from controllers.web.wraps import WebApiResource
from extensions.ext_database import db
from fields.file_fields import FileResponse
from libs.helper import dump_response
from models.model import App, EndUser
from services.file_service import FileService
@@ -85,4 +84,5 @@ class FileApi(WebApiResource):
except services.errors.file.UnsupportedFileTypeError:
raise UnsupportedFileTypeError()
return dump_response(FileResponse, upload_file), 201
response = FileResponse.model_validate(upload_file, from_attributes=True)
return response.model_dump(mode="json"), 201
+2 -2
View File
@@ -69,7 +69,7 @@ class ForgotPasswordSendEmailApi(Resource):
else:
language = "en-US"
account = AccountService.get_account_by_email_with_case_fallback(request_email, session=db.session())
account = AccountService.get_account_by_email_with_case_fallback(db.session(), request_email)
if account is None:
raise AuthenticationFailedError()
else:
@@ -168,7 +168,7 @@ class ForgotPasswordResetApi(Resource):
email = reset_data.get("email", "")
account = AccountService.get_account_by_email_with_case_fallback(email, session=db.session())
account = AccountService.get_account_by_email_with_case_fallback(db.session(), email)
if account:
account = db.session.merge(account)
@@ -29,7 +29,6 @@ from extensions.ext_database import db
from fields.file_fields import FileResponse, FileWithSignedUrl
from graphon.file import helpers as file_helpers
from libs.exception import BaseHTTPException
from libs.helper import dump_response
from repositories.factory import DifyAPIRepositoryFactory
from services.file_service import FileService
from services.human_input_file_upload_service import (
@@ -142,7 +141,8 @@ def _upload_local_file(context):
except services.errors.file.BlockedFileExtensionError as exc:
raise BlockedFileExtensionError() from exc
return upload_file.id, dump_response(FileResponse, upload_file)
response = FileResponse.model_validate(upload_file, from_attributes=True)
return upload_file.id, response
def _upload_remote_file(context, url: str):
@@ -186,7 +186,7 @@ def _upload_remote_file(context, url: str):
created_by=upload_file.created_by,
created_at=int(upload_file.created_at.timestamp()),
)
return upload_file.id, response.model_dump(mode="json")
return upload_file.id, response
@web_ns.route("/human-input-forms/files")
@@ -209,5 +209,4 @@ class HumanInputFileUploadApi(Resource):
file_id, response = _upload_local_file(context=context)
upload_service.record_upload_file(context=context, file_id=file_id)
# response-contract:ignore pre-dumped response. See above
return response, 201
return response.model_dump(mode="json"), 201
+58 -109
View File
@@ -2,12 +2,14 @@
Web App Human Input Form APIs.
"""
import json
import logging
from collections.abc import Sequence
from typing import Self
from typing import Any, NotRequired, TypedDict
from flask import request
from flask import Response, request
from flask_restx import Resource
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import select
from sqlalchemy.orm import sessionmaker
from werkzeug.exceptions import Forbidden
@@ -18,58 +20,35 @@ from controllers.common.human_input import HumanInputFormSubmitPayload, stringif
from controllers.common.schema import register_response_schema_models, register_schema_models
from controllers.web import web_ns
from controllers.web.error import WebFormRateLimitExceededError
from controllers.web.site import WebAppSiteResponse
from core.workflow.nodes.human_input.entities import FormInputConfig, UserActionConfig
from controllers.web.site import serialize_app_site_payload
from core.workflow.nodes.human_input.entities import FormInputConfig
from extensions.ext_database import db
from fields.base import ResponseModel
from libs.helper import RateLimiter, dump_response, extract_remote_ip, to_timestamp
from libs.helper import RateLimiter, extract_remote_ip, to_timestamp
from models.account import TenantStatus
from models.model import App, Site
from repositories.factory import DifyAPIRepositoryFactory
from services.feature_service import FeatureService
from services.human_input_file_upload_service import HumanInputFileUploadService
from services.human_input_service import Form, FormNotFoundError, HumanInputService
logger = logging.getLogger(__name__)
class HumanInputUploadTokenResponse(ResponseModel):
class HumanInputUploadTokenResponse(BaseModel):
upload_token: str
expires_at: int
class HumanInputFormDefinitionResponse(ResponseModel):
form_content: str
inputs: list[FormInputConfig]
class HumanInputFormDefinitionResponse(BaseModel):
form_content: Any
inputs: Any
resolved_default_values: dict[str, str]
user_actions: list[UserActionConfig]
user_actions: Any
expiration_time: int
site: WebAppSiteResponse | None = None
@classmethod
def from_form(
cls,
form: Form,
*,
inputs: Sequence[FormInputConfig] = (),
site: WebAppSiteResponse | None = None,
) -> Self:
definition_payload = form.get_definition().model_dump(mode="json")
expiration_time = to_timestamp(form.expiration_time)
if expiration_time is None:
raise ValueError("Human input form expiration_time is required")
return cls(
form_content=definition_payload["rendered_content"],
inputs=list(inputs),
resolved_default_values=stringify_form_default_values(definition_payload["default_values"]),
user_actions=definition_payload["user_actions"],
expiration_time=expiration_time,
site=site,
)
site: dict[str, Any] | None = Field(default=None)
class HumanInputFormSubmitResponse(ResponseModel):
pass
class HumanInputFormSubmitResponse(BaseModel):
model_config = ConfigDict(extra="forbid")
register_schema_models(web_ns, HumanInputFormSubmitPayload)
@@ -107,26 +86,40 @@ def _create_upload_service() -> HumanInputFileUploadService:
)
class FormDefinitionPayload(TypedDict):
form_content: Any
inputs: Any
resolved_default_values: dict[str, str]
user_actions: Any
expiration_time: int
site: NotRequired[dict]
def _jsonify_form_definition(
form: Form,
*,
inputs: Sequence[FormInputConfig] = (),
site_payload: dict | None = None,
) -> Response:
"""Return the form payload (optionally with site) as a JSON response."""
definition_payload = form.get_definition().model_dump(mode="json")
payload: FormDefinitionPayload = {
"form_content": definition_payload["rendered_content"],
"inputs": [i.model_dump(mode="json") for i in inputs],
"resolved_default_values": stringify_form_default_values(definition_payload["default_values"]),
"user_actions": definition_payload["user_actions"],
"expiration_time": to_timestamp(form.expiration_time),
}
if site_payload is not None:
payload["site"] = site_payload
return Response(json.dumps(payload, ensure_ascii=False), mimetype="application/json")
@web_ns.route("/form/human_input/<string:form_token>/upload-token")
class HumanInputFormUploadTokenApi(Resource):
"""API for issuing HITL upload tokens for active human input forms."""
@web_ns.doc("create_human_input_form_upload_token")
@web_ns.doc(description="Issue an upload token for an active human input form")
@web_ns.doc(params={"form_token": "Human input form token"})
@web_ns.doc(
responses={
200: "Upload token issued successfully",
404: "Form not found",
412: "Form already submitted or expired",
429: "Too many requests",
}
)
@web_ns.response(
200,
"Upload token issued successfully",
web_ns.models[HumanInputUploadTokenResponse.__name__],
)
@web_ns.response(200, "Success", web_ns.models[HumanInputUploadTokenResponse.__name__])
def post(self, form_token: str):
"""
Issue an upload token for a human input form.
@@ -143,9 +136,11 @@ class HumanInputFormUploadTokenApi(Resource):
except FormNotFoundError:
raise NotFoundError("Form not found")
return HumanInputUploadTokenResponse(
upload_token=token.upload_token, expires_at=to_timestamp(token.expires_at)
).model_dump(mode="json"), 200
response = HumanInputUploadTokenResponse(
upload_token=token.upload_token,
expires_at=to_timestamp(token.expires_at),
)
return response.model_dump(mode="json"), 200
@web_ns.route("/form/human_input/<string:form_token>")
@@ -155,23 +150,7 @@ class HumanInputFormApi(Resource):
# NOTE(QuantumGhost): this endpoint is unauthenticated on purpose for now.
# def get(self, _app_model: App, _end_user: EndUser, form_token: str):
@web_ns.doc("get_human_input_form")
@web_ns.doc(description="Get a human input form definition by token")
@web_ns.doc(params={"form_token": "Human input form token"})
@web_ns.doc(
responses={
200: "Form retrieved successfully",
403: "Forbidden",
404: "Form not found",
412: "Form already submitted or expired",
429: "Too many requests",
}
)
@web_ns.response(
200,
"Form retrieved successfully",
web_ns.models[HumanInputFormDefinitionResponse.__name__],
)
@web_ns.response(200, "Success", web_ns.models[HumanInputFormDefinitionResponse.__name__])
def get(self, form_token: str):
"""
Get human input form definition by token.
@@ -193,47 +172,17 @@ class HumanInputFormApi(Resource):
service.ensure_form_active(form)
app_model, site = _get_app_site_from_form(form)
tenant = app_model.tenant
if tenant is None:
raise Forbidden()
inputs = service.resolve_form_inputs(form)
features = FeatureService.get_features(app_model.tenant_id, exclude_vector_space=True)
return dump_response(
HumanInputFormDefinitionResponse,
HumanInputFormDefinitionResponse.from_form(
form,
inputs=inputs,
site=WebAppSiteResponse.from_app_site(
tenant=tenant,
app_model=app_model,
site=site,
end_user_id=None,
features=features,
can_replace_logo=features.can_replace_logo,
),
),
return _jsonify_form_definition(
form,
inputs=inputs,
site_payload=serialize_app_site_payload(app_model, site, None),
)
# def post(self, _app_model: App, _end_user: EndUser, form_token: str):
@web_ns.response(200, "Success", web_ns.models[HumanInputFormSubmitResponse.__name__])
@web_ns.expect(web_ns.models[HumanInputFormSubmitPayload.__name__])
@web_ns.doc("submit_human_input_form")
@web_ns.doc(description="Submit a human input form by token")
@web_ns.doc(params={"form_token": "Human input form token"})
@web_ns.doc(
responses={
200: "Form submitted successfully",
400: "Bad request - invalid submission data",
404: "Form not found",
412: "Form already submitted or expired",
429: "Too many requests",
}
)
@web_ns.response(
200,
"Form submitted successfully",
web_ns.models[HumanInputFormSubmitResponse.__name__],
)
def post(self, form_token: str):
"""
Submit human input form by token.
@@ -276,7 +225,7 @@ class HumanInputFormApi(Resource):
except FormNotFoundError:
raise NotFoundError("Form not found")
return HumanInputFormSubmitResponse().model_dump(mode="json"), 200
return {}, 200
def _get_app_site_from_form(form: Form) -> tuple[App, Site]:
@@ -289,7 +238,7 @@ def _get_app_site_from_form(form: Form) -> tuple[App, Site]:
if site is None:
raise Forbidden()
if app_model.tenant is None or app_model.tenant.status == TenantStatus.ARCHIVE:
if app_model.tenant and app_model.tenant.status == TenantStatus.ARCHIVE:
raise Forbidden()
return app_model, site
+1 -1
View File
@@ -147,7 +147,7 @@ class LoginStatusApi(Resource):
return LoginStatusResponse(logged_in=bool(token), app_logged_in=False).model_dump(mode="json")
app_id = AppService.get_app_id_by_code(app_code, session=db.session())
is_public = not dify_config.ENTERPRISE_ENABLED or not WebAppAuthService.is_app_require_permission_check(
app_id=app_id, session=db.session()
db.session(), app_id=app_id
)
user_logged_in = False
-1
View File
@@ -188,7 +188,6 @@ class MessageMoreLikeThisApi(WebApiResource):
streaming=streaming,
)
# response-contract:ignore compact_generate_response
return helper.compact_generate_response(response)
except MessageNotExistsError:
raise NotFound("Message Not Exists.")
+1 -1
View File
@@ -62,7 +62,7 @@ class PassportResource(Resource):
raise Unauthorized("X-App-Code header is missing.")
if system_features.webapp_auth.enabled:
enterprise_user_decoded = decode_enterprise_webapp_user_id(access_token)
app_auth_type = WebAppAuthService.get_app_auth_type(app_code=app_code, session=db.session())
app_auth_type = WebAppAuthService.get_app_auth_type(db.session(), app_code=app_code)
if app_auth_type != WebAppAuthType.PUBLIC:
if not enterprise_user_decoded:
raise WebAppAuthRequiredError()
+6 -4
View File
@@ -65,10 +65,11 @@ class RemoteFileInfoApi(WebApiResource):
# failed back to get method
resp = remote_fetcher.make_request("GET", decoded_url, timeout=3)
resp.raise_for_status()
return RemoteFileInfo(
info = RemoteFileInfo(
file_type=resp.headers.get("Content-Type", "application/octet-stream"),
file_length=int(resp.headers.get("Content-Length", -1)),
).model_dump(mode="json")
)
return info.model_dump(mode="json")
@web_ns.route("/remote-files/upload")
@@ -140,7 +141,7 @@ class RemoteFileUploadApi(WebApiResource):
except services.errors.file.UnsupportedFileTypeError:
raise UnsupportedFileTypeError
return FileWithSignedUrl(
payload1 = FileWithSignedUrl(
id=upload_file.id,
name=upload_file.name,
size=upload_file.size,
@@ -149,4 +150,5 @@ class RemoteFileUploadApi(WebApiResource):
mime_type=upload_file.mime_type,
created_by=upload_file.created_by,
created_at=int(upload_file.created_at.timestamp()),
).model_dump(mode="json"), 201
)
return payload1.model_dump(mode="json"), 201
+6 -5
View File
@@ -44,12 +44,14 @@ class SavedMessageListApi(WebApiResource):
query = SavedMessageListQuery.model_validate(raw_args)
pagination = SavedMessageService.pagination_by_last_id(
app_model, end_user, query.last_id, query.limit, session=db.session()
db.session(), app_model, end_user, query.last_id, query.limit
)
adapter = TypeAdapter(SavedMessageItem)
items = [adapter.validate_python(message, from_attributes=True) for message in pagination.data]
return SavedMessageInfiniteScrollPagination(
limit=pagination.limit, has_more=pagination.has_more, data=items
limit=pagination.limit,
has_more=pagination.has_more,
data=items,
).model_dump(mode="json")
@web_ns.doc("Save Message")
@@ -78,7 +80,7 @@ class SavedMessageListApi(WebApiResource):
payload = SavedMessageCreatePayload.model_validate(web_ns.payload or {})
try:
SavedMessageService.save(app_model, end_user, payload.message_id, session=db.session())
SavedMessageService.save(db.session(), app_model, end_user, payload.message_id)
except MessageNotExistsError:
raise NotFound("Message Not Exists.")
@@ -100,13 +102,12 @@ class SavedMessageApi(WebApiResource):
500: "Internal Server Error",
}
)
@web_ns.response(204, "Message removed successfully")
def delete(self, app_model: App, end_user: EndUser, message_id: UUID):
message_id_str = str(message_id)
if app_model.mode != "completion":
raise NotCompletionAppError()
SavedMessageService.delete(app_model, end_user, message_id_str, session=db.session())
SavedMessageService.delete(db.session(), app_model, end_user, message_id_str)
return "", 204
+132 -100
View File
@@ -1,6 +1,7 @@
from typing import Any, Self
from typing import Any, cast
from pydantic import AliasChoices, Field, computed_field
from flask_restx import fields, marshal, marshal_with
from pydantic import Field
from sqlalchemy import select
from werkzeug.exceptions import Forbidden
@@ -10,19 +11,30 @@ from controllers.web import web_ns
from controllers.web.wraps import WebApiResource
from extensions.ext_database import db
from fields.base import ResponseModel
from libs.helper import build_icon_url
from models.account import Tenant, TenantStatus
from libs.helper import AppIconUrlField
from models.account import TenantStatus
from models.model import App, EndUser, Site
from services.feature_service import FeatureModel, FeatureService
class WebSiteResponse(ResponseModel):
title: str
class AppSiteModelConfigResponse(ResponseModel):
opening_statement: str | None = None
suggested_questions: Any
suggested_questions_after_answer: Any
more_like_this: Any
model: Any
user_input_form: Any
pre_prompt: str | None = None
class AppSiteResponse(ResponseModel):
title: str | None = None
chat_color_theme: str | None = None
chat_color_theme_inverted: bool
chat_color_theme_inverted: bool | None = None
icon_type: str | None = None
icon: str | None = None
icon_background: str | None = None
icon_url: str | None = None
description: str | None = None
copyright: str | None = None
privacy_policy: str | None = None
@@ -33,98 +45,65 @@ class WebSiteResponse(ResponseModel):
show_workflow_steps: bool | None = None
use_icon_as_answer_icon: bool | None = None
@computed_field(return_type=str | None) # type: ignore[prop-decorator]
@property
def icon_url(self) -> str | None:
return build_icon_url(self.icon_type, self.icon)
class WebModelConfigResponse(ResponseModel):
opening_statement: str | None = None
suggested_questions: Any = Field(
default=None,
validation_alias=AliasChoices("suggested_questions_list", "suggested_questions"),
)
suggested_questions_after_answer: Any = Field(
default=None,
validation_alias=AliasChoices("suggested_questions_after_answer_dict", "suggested_questions_after_answer"),
)
more_like_this: Any = Field(
default=None,
validation_alias=AliasChoices("more_like_this_dict", "more_like_this"),
)
model: Any = Field(default=None, validation_alias=AliasChoices("model_dict", "model"))
user_input_form: Any = Field(
default=None,
validation_alias=AliasChoices("user_input_form_list", "user_input_form"),
)
pre_prompt: str | None = None
class WebAppCustomConfigResponse(ResponseModel):
remove_webapp_brand: bool
replace_webapp_logo: str | None = None
class WebAppSiteResponse(ResponseModel):
class AppSiteInfoResponse(ResponseModel):
app_id: str
end_user_id: str | None = None
enable_site: bool
site: WebSiteResponse
model_config_: WebModelConfigResponse | None = Field(
default=None, validation_alias="model_config", serialization_alias="model_config"
)
plan: str
site: AppSiteResponse
model_config_: AppSiteModelConfigResponse | None = Field(default=None, alias="model_config")
plan: str | None = None
can_replace_logo: bool
custom_config: WebAppCustomConfigResponse | None = None
@classmethod
def from_app_site(
cls,
*,
tenant: Tenant,
app_model: App,
site: Site,
end_user_id: str | None,
features: FeatureModel,
can_replace_logo: bool,
) -> Self:
custom_config = None
if can_replace_logo:
replace_webapp_logo = (
f"{dify_config.FILES_URL}/files/workspaces/{tenant.id}/webapp-logo"
if tenant.custom_config_dict.get("replace_webapp_logo")
else None
)
custom_config = WebAppCustomConfigResponse(
remove_webapp_brand=tenant.custom_config_dict.get("remove_webapp_brand", False),
replace_webapp_logo=replace_webapp_logo,
)
site_response = WebSiteResponse.model_validate(site, from_attributes=True)
if features.billing.enabled and not features.webapp_copyright_enabled:
site_response.copyright = None
site_response.input_placeholder = None
return cls(
app_id=app_model.id,
end_user_id=end_user_id,
enable_site=app_model.enable_site,
site=site_response,
model_config_=None,
plan=tenant.plan,
can_replace_logo=can_replace_logo,
custom_config=custom_config,
)
custom_config: dict[str, Any] | None = Field(default=None)
register_response_schema_models(
web_ns, WebSiteResponse, WebModelConfigResponse, WebAppCustomConfigResponse, WebAppSiteResponse
)
register_response_schema_models(web_ns, AppSiteInfoResponse)
@web_ns.route("/site")
class AppSiteApi(WebApiResource):
"""Resource for app sites."""
model_config_fields = {
"opening_statement": fields.String,
"suggested_questions": fields.Raw(attribute="suggested_questions_list"),
"suggested_questions_after_answer": fields.Raw(attribute="suggested_questions_after_answer_dict"),
"more_like_this": fields.Raw(attribute="more_like_this_dict"),
"model": fields.Raw(attribute="model_dict"),
"user_input_form": fields.Raw(attribute="user_input_form_list"),
"pre_prompt": fields.String,
}
site_fields = {
"title": fields.String,
"chat_color_theme": fields.String,
"chat_color_theme_inverted": fields.Boolean,
"icon_type": fields.String,
"icon": fields.String,
"icon_background": fields.String,
"icon_url": AppIconUrlField,
"description": fields.String,
"copyright": fields.String,
"privacy_policy": fields.String,
"input_placeholder": fields.String,
"custom_disclaimer": fields.String,
"default_language": fields.String,
"prompt_public": fields.Boolean,
"show_workflow_steps": fields.Boolean,
"use_icon_as_answer_icon": fields.Boolean,
}
app_fields = {
"app_id": fields.String,
"end_user_id": fields.String,
"enable_site": fields.Boolean,
"site": fields.Nested(site_fields),
"model_config": fields.Nested(model_config_fields, allow_null=True),
"plan": fields.String,
"can_replace_logo": fields.Boolean,
"custom_config": fields.Raw(attribute="custom_config"),
}
@web_ns.doc("Get App Site Info")
@web_ns.doc(description="Retrieve app site information and configuration.")
@web_ns.doc(
@@ -137,26 +116,79 @@ class AppSiteApi(WebApiResource):
500: "Internal Server Error",
}
)
@web_ns.response(200, "Success", web_ns.models[WebAppSiteResponse.__name__])
@web_ns.response(200, "Success", web_ns.models[AppSiteInfoResponse.__name__])
@marshal_with(app_fields)
def get(self, app_model: App, end_user: EndUser):
"""Retrieve app site info."""
# get site
site = db.session.scalar(select(Site).where(Site.app_id == app_model.id).limit(1))
if site is None:
if not site:
raise Forbidden()
tenant = app_model.tenant
if tenant is None or tenant.status == TenantStatus.ARCHIVE:
if app_model.tenant and app_model.tenant.status == TenantStatus.ARCHIVE:
raise Forbidden()
features = FeatureService.get_features(app_model.tenant_id, exclude_vector_space=True)
return WebAppSiteResponse.from_app_site(
tenant=tenant,
app_model=app_model,
site=site,
end_user_id=end_user.id,
features=features,
can_replace_logo=features.can_replace_logo,
).model_dump(mode="json")
return AppSiteInfo(
app_model.tenant,
app_model,
serialize_runtime_site(site, features),
end_user.id,
features.can_replace_logo,
)
class AppSiteInfo:
"""Class to store site information."""
def __init__(self, tenant, app, site, end_user, can_replace_logo):
"""Initialize AppSiteInfo instance."""
self.app_id = app.id
self.end_user_id = end_user
self.enable_site = app.enable_site
self.site = site
self.model_config = None
self.plan = tenant.plan
self.can_replace_logo = can_replace_logo
if can_replace_logo:
base_url = dify_config.FILES_URL
remove_webapp_brand = tenant.custom_config_dict.get("remove_webapp_brand", False)
replace_webapp_logo = (
f"{base_url}/files/workspaces/{tenant.id}/webapp-logo"
if tenant.custom_config_dict.get("replace_webapp_logo")
else None
)
self.custom_config = {
"remove_webapp_brand": remove_webapp_brand,
"replace_webapp_logo": replace_webapp_logo,
}
def serialize_site(site: Site) -> dict[str, Any]:
"""Serialize Site model using the same schema as AppSiteApi."""
return cast(dict[str, Any], marshal(site, AppSiteApi.site_fields))
def serialize_runtime_site(site: Site, features: FeatureModel) -> dict[str, Any]:
site_payload = serialize_site(site)
if not features.billing.enabled or features.webapp_copyright_enabled:
return site_payload
site_payload["copyright"] = None
site_payload["input_placeholder"] = None
return site_payload
def serialize_app_site_payload(app_model: App, site: Site, end_user_id: str | None) -> dict[str, Any]:
features = FeatureService.get_features(app_model.tenant_id, exclude_vector_space=True)
app_site_info = AppSiteInfo(
app_model.tenant,
app_model,
serialize_runtime_site(site, features),
end_user_id,
features.can_replace_logo,
)
return cast(dict[str, Any], marshal(app_site_info, AppSiteApi.app_fields))
+1 -2
View File
@@ -76,7 +76,6 @@ class WorkflowRunApi(WebApiResource):
streaming=True,
)
# response-contract:ignore compact_generate_response
return helper.compact_generate_response(response)
except ProviderTokenNotInitError as ex:
raise ProviderNotInitializeError(ex.description)
@@ -130,4 +129,4 @@ class WorkflowTaskStopApi(WebApiResource):
# New graph engine command channel mechanism
GraphEngineManager(redis_client).send_stop_command(task_id)
return SimpleResultResponse(result="success").model_dump(mode="json")
return {"result": "success"}
+1 -3
View File
@@ -129,9 +129,7 @@ def _validate_user_accessibility(
if not webapp_settings:
raise WebAppAuthRequiredError("Web app settings not found.")
if WebAppAuthService.is_app_require_permission_check(
access_mode=webapp_settings.access_mode, session=db.session()
):
if WebAppAuthService.is_app_require_permission_check(db.session(), access_mode=webapp_settings.access_mode):
app_id = AppService.get_app_id_by_code(app_code, session=db.session())
if not EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp(user_id, app_id):
raise WebAppAuthAccessDeniedError()
@@ -48,7 +48,6 @@ from core.repositories import DifyCoreRepositoryFactory
from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
from extensions.ext_database import db
from factories import file_factory
from graphon.filters import ResponseStreamFilter
from graphon.graph_engine.layers import GraphEngineLayer
from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
from graphon.runtime import GraphRuntimeState
@@ -270,7 +269,6 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
workflow_node_execution_repository: WorkflowNodeExecutionRepository,
graph_runtime_state: GraphRuntimeState,
pause_state_config: PauseStateLayerConfig | None = None,
response_stream_filter: ResponseStreamFilter | None = None,
):
"""
Resume a paused advanced chat execution.
@@ -300,7 +298,6 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
stream=application_generate_entity.stream,
pause_state_config=pause_state_config,
graph_runtime_state=graph_runtime_state,
response_stream_filter=response_stream_filter,
)
def single_iteration_generate(
@@ -495,7 +492,6 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
pause_state_config: PauseStateLayerConfig | None = None,
graph_runtime_state: GraphRuntimeState | None = None,
graph_engine_layers: Sequence[GraphEngineLayer] = (),
response_stream_filter: ResponseStreamFilter | None = None,
) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], None, None]:
"""
Generate App response.
@@ -543,14 +539,12 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
)
graph_layers: list[GraphEngineLayer] = list(graph_engine_layers)
resolved_response_stream_filter = response_stream_filter or ResponseStreamFilter()
if pause_state_config is not None:
graph_layers.append(
PauseStatePersistenceLayer(
session_factory=pause_state_config.session_factory,
generate_entity=application_generate_entity,
state_owner_user_id=pause_state_config.state_owner_user_id,
response_stream_filter=resolved_response_stream_filter,
)
)
@@ -571,7 +565,6 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
"workflow_node_execution_repository": workflow_node_execution_repository,
"graph_engine_layers": tuple(graph_layers),
"graph_runtime_state": graph_runtime_state,
"response_stream_filter": resolved_response_stream_filter,
},
)
@@ -611,7 +604,6 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
workflow_node_execution_repository: WorkflowNodeExecutionRepository,
graph_engine_layers: Sequence[GraphEngineLayer] = (),
graph_runtime_state: GraphRuntimeState | None = None,
response_stream_filter: ResponseStreamFilter | None = None,
):
"""
Generate worker in a new thread.
@@ -671,7 +663,6 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
workflow_node_execution_repository=workflow_node_execution_repository,
graph_engine_layers=graph_engine_layers,
graph_runtime_state=graph_runtime_state,
response_stream_filter=response_stream_filter,
)
try:
@@ -44,7 +44,6 @@ from extensions.ext_redis import redis_client
from extensions.otel import WorkflowAppRunnerHandler, trace_span
from extensions.workflow_warm_shutdown import WORKFLOW_WARM_SHUTDOWN_ABORT_REASON, celery_warm_shutdown_started
from graphon.enums import WorkflowType
from graphon.filters import ResponseStreamFilter
from graphon.graph_engine.command_channels import RedisChannel
from graphon.graph_engine.layers import GraphEngineLayer
from graphon.runtime import GraphRuntimeState, VariablePool
@@ -79,7 +78,6 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
workflow_node_execution_repository: WorkflowNodeExecutionRepository,
graph_engine_layers: Sequence[GraphEngineLayer] = (),
graph_runtime_state: GraphRuntimeState | None = None,
response_stream_filter: ResponseStreamFilter | None = None,
):
super().__init__(
queue_manager=queue_manager,
@@ -97,7 +95,6 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
self._workflow_execution_repository = workflow_execution_repository
self._workflow_node_execution_repository = workflow_node_execution_repository
self._resume_graph_runtime_state = graph_runtime_state
self._response_stream_filter = response_stream_filter
@trace_span(WorkflowAppRunnerHandler)
def run(self):
@@ -244,7 +241,6 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
variable_pool=variable_pool,
graph_runtime_state=graph_runtime_state,
command_channel=command_channel,
response_stream_filter=self._response_stream_filter,
)
self._queue_manager.graph_runtime_state = graph_runtime_state
+89 -81
View File
@@ -1,11 +1,14 @@
"""Agent App generator: orchestrate Agent App chat and finalize executions.
Agent App turns mirror the agent_chat generator (conversation + message +
The primary mode mirrors the agent_chat generator (conversation + message +
queue + streamed response over the EasyUI chat pipeline), but the backing
config comes from the bound Agent Soul and the answer is produced by
``AgentAppRunner`` calling the dify-agent backend rather than an in-process
LLM/ReAct loop. Build-chat finalization uses this same streamed path and only
changes the runtime exit policy carried to the backend.
LLM/ReAct loop.
It also exposes a stateless build-finalize mode that reuses existing runtime
context from the bound debug conversation, triggers the Agent backend side
effect synchronously, and skips Dify-side chat/message persistence.
"""
from __future__ import annotations
@@ -38,16 +41,13 @@ from core.app.apps.exc import GenerateTaskStoppedError
from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
from core.app.entities.app_invoke_entities import (
AGENT_RUNTIME_EXIT_INTENT_ARG,
AgentAppGenerateEntity,
AgentRuntimeExitIntent,
DifyRunContext,
InvokeFrom,
UserFrom,
)
from core.app.llm.model_access import build_dify_model_access
from core.ops.ops_trace_manager import TraceQueueManager
from core.workflow.file_reference import build_file_reference, is_canonical_file_reference
from extensions.ext_database import db
from models import Account, App, EndUser, Message
from models.agent import (
@@ -64,68 +64,12 @@ from services.conversation_service import ConversationService
logger = logging.getLogger(__name__)
_REFERENCE_FILE_TRANSFER_METHODS = {"local_file", "tool_file", "datasource_file"}
def _append_prompt_file_mappings(query: str, prompt_file_mappings: Sequence[JsonValue]) -> str:
"""Append labeled, prompt-safe file locators to the backend user prompt."""
prompt_files = _prompt_file_locators(prompt_file_mappings)
if not prompt_files:
"""Append raw request file references to the backend user prompt."""
if not prompt_file_mappings:
return query
payload = json.dumps(prompt_files, ensure_ascii=False, separators=(",", ":"))
return (
f"{query}\n"
"User provided files: use dify-agent file download with the listed transfer_method and reference/url "
"to get the files and investigate them\n"
f"{payload}"
)
def _prompt_file_locators(prompt_file_mappings: Sequence[JsonValue]) -> list[dict[str, str]]:
locators: list[dict[str, str]] = []
for file_mapping in prompt_file_mappings:
if not isinstance(file_mapping, Mapping):
continue
locator = _prompt_file_locator(file_mapping)
if locator is not None:
locators.append(locator)
return locators
def _prompt_file_locator(file_mapping: Mapping[str, object]) -> dict[str, str] | None:
transfer_method = _string_value(file_mapping, "transfer_method")
if transfer_method == "remote_url":
url = _string_value(file_mapping, "url") or _string_value(file_mapping, "remote_url")
if url is None:
return None
return {"transfer_method": "remote_url", "url": url}
elif transfer_method in _REFERENCE_FILE_TRANSFER_METHODS:
if transfer_method is None:
return None
reference = _canonical_file_reference(
_string_value(file_mapping, "reference")
or _string_value(file_mapping, "upload_file_id")
or _string_value(file_mapping, "file_id")
or _string_value(file_mapping, "id")
)
if reference is None:
return None
return {"transfer_method": transfer_method, "reference": reference}
else:
return None
def _canonical_file_reference(reference: str | None) -> str | None:
if reference is None:
return None
if reference.startswith("dify-file-ref:"):
return reference if is_canonical_file_reference(reference) else None
return build_file_reference(record_id=reference)
def _string_value(mapping: Mapping[str, object], key: str) -> str | None:
value = mapping.get(key)
return value if isinstance(value, str) and value else None
return f"{query}\n{json.dumps(list(prompt_file_mappings), ensure_ascii=False)}"
class AgentAppGenerator(MessageBasedAppGenerator):
@@ -176,7 +120,6 @@ class AgentAppGenerator(MessageBasedAppGenerator):
model_conf = ModelConfigConverter.convert(app_config)
trace_manager = TraceQueueManager(app_model.id, user.id if isinstance(user, Account) else user.session_id)
agent_runtime_exit_intent = self._resolve_agent_runtime_exit_intent(args)
application_generate_entity = AgentAppGenerateEntity(
task_id=str(uuid.uuid4()),
@@ -206,7 +149,6 @@ class AgentAppGenerator(MessageBasedAppGenerator):
agent_config_snapshot_id=agent_config_id,
agent_config_version_kind=agent_config_version_kind,
agent_runtime_session_snapshot_id=runtime_session_snapshot_id,
agent_runtime_exit_intent=agent_runtime_exit_intent,
)
conversation, message = self._init_generate_records(application_generate_entity, conversation)
@@ -245,6 +187,86 @@ class AgentAppGenerator(MessageBasedAppGenerator):
)
return AgentAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
def generate_stateless(
self,
*,
app_model: App,
user: Account | EndUser,
args: Mapping[str, Any],
invoke_from: InvokeFrom,
) -> Mapping[str, Any]:
"""Run one Agent App turn without persisting Dify conversation messages."""
query = self._require_query(args)
conversation_id = args.get("conversation_id")
if not isinstance(conversation_id, str) or not conversation_id:
raise AgentAppGeneratorError("conversation_id is required")
agent, agent_config_id, agent_config_version_kind, agent_soul = self._resolve_agent(
app_model,
invoke_from=invoke_from,
draft_type=args.get("draft_type"),
user=user,
)
runtime_session_snapshot_id = self._runtime_session_snapshot_id(
invoke_from=invoke_from,
snapshot_id=agent_config_id,
)
return self._run_stateless(
app_model=app_model,
user=user,
invoke_from=invoke_from,
query=query,
conversation_id=conversation_id,
agent=agent,
agent_config_id=agent_config_id,
agent_config_version_kind=agent_config_version_kind,
agent_soul=agent_soul,
runtime_session_snapshot_id=runtime_session_snapshot_id,
)
def _run_stateless(
self,
*,
app_model: App,
user: Account | EndUser,
invoke_from: InvokeFrom,
query: str,
conversation_id: str,
agent: Agent,
agent_config_id: str,
agent_config_version_kind: Literal["snapshot", "draft", "build_draft"],
agent_soul: AgentSoulConfig,
runtime_session_snapshot_id: str | None,
) -> Mapping[str, Any]:
"""Run the Agent backend without creating or updating Dify chat records.
Build-chat finalization is an action against the Agent backend (for
example, ``dify-agent config push``). It may reuse the active build-chat
runtime snapshot for shell/config context, but the API side must not add
a synthetic user/assistant turn to the debug conversation.
"""
dify_context = DifyRunContext(
tenant_id=app_model.tenant_id,
app_id=app_model.id,
user_id=user.id,
user_from=UserFrom.ACCOUNT if isinstance(user, Account) else UserFrom.END_USER,
invoke_from=invoke_from,
)
self._build_runner(dify_context).run_stateless(
dify_context=dify_context,
agent_id=agent.id,
agent_config_snapshot_id=agent_config_id,
agent_config_version_kind=agent_config_version_kind,
agent_soul=agent_soul,
conversation_id=conversation_id,
query=query,
idempotency_key=str(uuid.uuid4()),
session_scope_snapshot_id=runtime_session_snapshot_id,
)
return {"result": "success"}
def resume_after_form_submission(
self,
*,
@@ -333,7 +355,6 @@ class AgentAppGenerator(MessageBasedAppGenerator):
target=self._generate_worker,
kwargs={
"flask_app": current_app._get_current_object(), # type: ignore
"session": db.session(),
"context": context,
"application_generate_entity": application_generate_entity,
"queue_manager": queue_manager,
@@ -455,7 +476,6 @@ class AgentAppGenerator(MessageBasedAppGenerator):
model_name=application_generate_entity.model_conf.model,
queue_manager=queue_manager,
session_scope_snapshot_id=application_generate_entity.agent_runtime_session_snapshot_id,
agent_runtime_exit_intent=application_generate_entity.agent_runtime_exit_intent,
)
except GenerateTaskStoppedError:
pass
@@ -472,18 +492,6 @@ class AgentAppGenerator(MessageBasedAppGenerator):
raise AgentAppGeneratorError("query is required")
return query.replace("\x00", "")
@staticmethod
def _resolve_agent_runtime_exit_intent(args: Mapping[str, Any]) -> AgentRuntimeExitIntent:
"""Resolve API-internal runtime exit policy from controller-owned args.
Only the private controller-injected "delete" value changes behavior.
Normal chat and resume flows default/fallback to "suspend" so public
payloads and invalid internal values preserve existing semantics.
"""
if args.get(AGENT_RUNTIME_EXIT_INTENT_ARG) == "delete":
return "delete"
return "suspend"
@staticmethod
def _build_runner(dify_context: DifyRunContext) -> AgentAppRunner:
credentials_provider, _ = build_dify_model_access(dify_context)
+171 -328
View File
@@ -1,10 +1,15 @@
"""Agent App runner: drive Agent backend turns for chat and finalization flows.
"""Agent App runner: drive Agent backend turns for both chat and finalize flows.
Unlike the legacy ``AgentChatAppRunner`` (which runs an in-process ReAct loop),
this runner delegates to the Agent backend, consumes the streamed event flow,
republishes the assistant answer through the existing EasyUI chat task
pipeline, and then either saves or retires the conversation-owned runtime
session depending on the turn's exit policy.
this runner delegates to the Agent backend and supports two execution modes.
- Normal chat turns build the run request from the Agent Soul + conversation,
consume backend stream events, republish the assistant answer through the
existing EasyUI chat task pipeline, and save the conversation
``session_snapshot`` on success for multi-turn continuity (S3).
- Stateless build-finalize turns reuse any prior conversation snapshot only to
construct the backend request, wait synchronously for backend completion, and
intentionally do not persist Dify-side chat records or runtime-session state.
"""
from __future__ import annotations
@@ -21,18 +26,16 @@ from dify_agent.protocol import DeferredToolResultsPayload
from pydantic import JsonValue
from clients.agent_backend import (
AgentBackendAgentMessageDeltaInternalEvent,
AgentBackendDeferredToolCallInternalEvent,
AgentBackendError,
AgentBackendInternalEventType,
AgentBackendRunClient,
AgentBackendRunEventAdapter,
AgentBackendRunFailedInternalEvent,
AgentBackendRunSucceededInternalEvent,
AgentBackendStreamInternalEvent,
extract_runtime_layer_specs,
)
from clients.agent_backend.session_cleanup import AgentBackendSessionCleanupPayload
from configs import dify_config
from core.app.apps.agent_app.runtime_request_builder import (
AgentAppRuntimeBuildContext,
AgentAppRuntimeRequest,
@@ -45,31 +48,17 @@ from core.app.apps.agent_app.session_store import (
)
from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
from core.app.apps.exc import GenerateTaskStoppedError
from core.app.entities.app_invoke_entities import AgentRuntimeExitIntent, DifyRunContext
from core.app.entities.queue_entities import (
QueueAgentMessageEvent,
QueueAgentThoughtEvent,
QueueLLMChunkEvent,
QueueMessageEndEvent,
)
from core.app.entities.app_invoke_entities import DifyRunContext
from core.app.entities.queue_entities import QueueAgentThoughtEvent, QueueLLMChunkEvent, QueueMessageEndEvent
from core.repositories.human_input_repository import HumanInputFormRepository, HumanInputFormRepositoryImpl
from core.workflow.nodes.agent_v2.ask_human_hitl import AskHumanFormBuildError, create_ask_human_form
from core.workflow.nodes.agent_v2.ask_human_resume import build_deferred_tool_results, resolve_ask_human_form
from extensions.ext_database import db
from graphon.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
from graphon.model_runtime.entities.message_entities import AssistantPromptMessage, PromptMessage, UserPromptMessage
from graphon.model_runtime.errors.invoke import (
InvokeAuthorizationError,
InvokeBadRequestError,
InvokeConnectionError,
InvokeError,
InvokeRateLimitError,
InvokeServerUnavailableError,
)
from models.agent_config_entities import AgentSoulConfig
from models.enums import CreatorUserRole
from models.model import MessageAgentThought
from tasks.agent_backend_session_cleanup_task import cleanup_conversation_agent_runtime_session
logger = logging.getLogger(__name__)
@@ -80,22 +69,6 @@ class _DefaultSessionScopeSnapshotId:
_DEFAULT_SESSION_SCOPE_SNAPSHOT_ID = _DefaultSessionScopeSnapshotId()
_AGENT_BACKEND_INVOKE_ERROR_BY_REASON: Mapping[str, type[InvokeError]] = {
"InvokeAuthorizationError": InvokeAuthorizationError,
"InvokeBadRequestError": InvokeBadRequestError,
"CredentialsValidateFailedError": InvokeBadRequestError,
"InvokeConnectionError": InvokeConnectionError,
"InvokeRateLimitError": InvokeRateLimitError,
"InvokeServerUnavailableError": InvokeServerUnavailableError,
}
def _agent_backend_failure_to_exception(event: AgentBackendRunFailedInternalEvent) -> Exception:
err_cls = _AGENT_BACKEND_INVOKE_ERROR_BY_REASON.get(event.reason or "")
if err_cls is not None:
return err_cls(event.error)
return AgentBackendError(event.error or "Agent backend run did not complete successfully.")
def _prompt_messages_from_query(user_query: str | None) -> list[PromptMessage]:
if not user_query:
@@ -162,25 +135,6 @@ def publish_text_delta(
queue_manager.publish(QueueLLMChunkEvent(chunk=chunk), PublishFrom.APPLICATION_MANAGER)
def publish_agent_message_delta(
*,
queue_manager: AppQueueManager,
model_name: str,
delta: str,
user_query: str | None = None,
) -> None:
"""Publish one agent-process text delta through the EasyUI chat pipeline."""
if not delta:
return
prompt_messages = _prompt_messages_from_query(user_query)
chunk = LLMResultChunk(
model=model_name,
prompt_messages=prompt_messages,
delta=LLMResultChunkDelta(index=0, message=AssistantPromptMessage(content=delta)),
)
queue_manager.publish(QueueAgentMessageEvent(chunk=chunk), PublishFrom.APPLICATION_MANAGER)
def publish_message_end(
*,
queue_manager: AppQueueManager,
@@ -205,7 +159,7 @@ def publish_message_end(
class _TextDeltaDebouncer:
"""Batch independent model text deltas before agent-message SSE output."""
"""Batch assistant text deltas on stream-event boundaries for final SSE output."""
def __init__(self, *, debounce_seconds: float) -> None:
self._debounce_seconds = debounce_seconds
@@ -238,13 +192,7 @@ class _TextDeltaDebouncer:
class _AgentProcessRecorder:
"""Persist Agent v2 process streams through the legacy thought model.
Thinking and answer rows expose snapshot updates for contiguous model-text
segments. Tool events close currently open text segments so later model text
starts a fresh row instead of replaying content that was already streamed
before the tool.
"""
"""Persist Agent v2 thinking/tool process events through the legacy thought model."""
def __init__(
self,
@@ -258,7 +206,6 @@ class _AgentProcessRecorder:
self._queue_manager = queue_manager
self._next_position = 1
self._thinking_by_index: dict[int, str] = {}
self._answer_thought_id: str | None = None
self._tool_by_index: dict[int, str] = {}
self._tool_by_call_id: dict[str, str] = {}
self._open_tool_by_name: dict[str, set[str]] = {}
@@ -314,41 +261,6 @@ class _AgentProcessRecorder:
if part_kind in {"tool-return", "builtin-tool-return"}:
self._record_tool_return_part(part)
def append_answer_text(self, content_delta: str) -> None:
if not content_delta:
return
self._thinking_by_index.clear()
if self._answer_thought_id is None:
self._answer_thought_id = self._create_thought(answer=content_delta)
return
self._update_thought(self._answer_thought_id, answer_delta=content_delta)
def trim_answer_suffix(self, final_answer: str) -> None:
if not final_answer or self._answer_thought_id is None:
return
row = db.session.get(MessageAgentThought, self._answer_thought_id)
if row is None:
return
answer = row.answer or ""
overlap = _suffix_prefix_overlap_length(answer, final_answer)
if overlap == 0:
return
row.answer = answer[:-overlap]
if _is_empty_answer_only_thought(row):
db.session.delete(row)
self._answer_thought_id = None
db.session.commit()
return
db.session.commit()
self._queue_manager.publish(
QueueAgentThoughtEvent(agent_thought_id=self._answer_thought_id), PublishFrom.APPLICATION_MANAGER
)
def _handle_tool_call_event(self, data: dict[str, Any]) -> None:
part = data.get("part")
if isinstance(part, dict):
@@ -369,7 +281,6 @@ class _AgentProcessRecorder:
)
def _append_thinking(self, index: int, content_delta: str) -> None:
self._answer_thought_id = None
thought_id = self._thinking_by_index.get(index)
if thought_id is None:
thought_id = self._create_thought(thought=content_delta)
@@ -378,7 +289,6 @@ class _AgentProcessRecorder:
self._update_thought(thought_id, thought_delta=content_delta)
def _record_tool_call_delta(self, index: int, delta: dict[str, Any]) -> None:
self._close_thinking_segments()
tool_call_id = _string_or_none(delta.get("tool_call_id"))
tool_name = _string_or_none(delta.get("tool_name_delta"))
args_delta = delta.get("args_delta")
@@ -395,10 +305,8 @@ class _AgentProcessRecorder:
tool=tool_name,
tool_input_delta=_json_or_text(args_delta),
)
self._remember_tool_thought(index=index, tool_call_id=tool_call_id, tool_name=tool_name, thought_id=thought_id)
def _record_tool_call_part(self, index: int, part: dict[str, Any]) -> None:
self._close_thinking_segments()
tool_call_id = _string_or_none(part.get("tool_call_id"))
tool_name = _string_or_none(part.get("tool_name"))
thought_id = self._lookup_tool_thought(index=index, tool_call_id=tool_call_id)
@@ -414,10 +322,8 @@ class _AgentProcessRecorder:
tool=tool_name,
tool_input=_json_or_text(part.get("args")),
)
self._remember_tool_thought(index=index, tool_call_id=tool_call_id, tool_name=tool_name, thought_id=thought_id)
def _record_tool_return_part(self, part: dict[str, Any]) -> None:
self._close_thinking_segments()
tool_call_id = _string_or_none(part.get("tool_call_id"))
tool_name = _string_or_none(part.get("tool_name"))
content = part.get("content")
@@ -426,7 +332,6 @@ class _AgentProcessRecorder:
self._record_tool_observation(tool_call_id=tool_call_id, tool_name=tool_name, observation=content)
def _record_tool_observation(self, *, tool_call_id: str | None, tool_name: str | None, observation: Any) -> None:
self._close_thinking_segments()
thought_id = self._lookup_observation_thought(tool_call_id=tool_call_id, tool_name=tool_name)
if thought_id is None:
thought_id = self._create_thought(tool=tool_name)
@@ -437,15 +342,12 @@ class _AgentProcessRecorder:
def _lookup_tool_thought(self, *, index: int, tool_call_id: str | None) -> str | None:
if tool_call_id and tool_call_id in self._tool_by_call_id:
return self._tool_by_call_id[tool_call_id]
if index < 0:
return None
return self._tool_by_index.get(index)
def _remember_tool_thought(
self, *, index: int, tool_call_id: str | None, tool_name: str | None, thought_id: str
) -> None:
if index >= 0:
self._tool_by_index[index] = thought_id
self._tool_by_index[index] = thought_id
if tool_call_id:
self._tool_by_call_id[tool_call_id] = thought_id
if tool_name:
@@ -461,24 +363,11 @@ class _AgentProcessRecorder:
return None
def _mark_tool_observed(self, thought_id: str) -> None:
self._tool_by_index = {index: value for index, value in self._tool_by_index.items() if value != thought_id}
self._tool_by_call_id = {
tool_call_id: value for tool_call_id, value in self._tool_by_call_id.items() if value != thought_id
}
for open_thought_ids in self._open_tool_by_name.values():
open_thought_ids.discard(thought_id)
def _close_thinking_segments(self) -> None:
self._thinking_by_index.clear()
self._answer_thought_id = None
def _create_thought(
self,
*,
thought: str | None = None,
answer: str | None = None,
tool: str | None = None,
tool_input: str | None = None,
self, *, thought: str | None = None, tool: str | None = None, tool_input: str | None = None
) -> str:
row = MessageAgentThought(
message_id=self._message_id,
@@ -495,7 +384,7 @@ class _AgentProcessRecorder:
message_unit_price=Decimal(0),
message_price_unit=Decimal("0.001"),
message_files="",
answer=answer or "",
answer="",
answer_token=0,
answer_unit_price=Decimal(0),
answer_price_unit=Decimal("0.001"),
@@ -525,7 +414,6 @@ class _AgentProcessRecorder:
tool_input: str | None = None,
tool_input_delta: str | None = None,
observation: str | None = None,
answer_delta: str | None = None,
) -> None:
row = db.session.get(MessageAgentThought, thought_id)
if row is None:
@@ -542,8 +430,6 @@ class _AgentProcessRecorder:
row.tool_input = f"{row.tool_input or ''}{tool_input_delta}"
if observation is not None:
row.observation = observation
if answer_delta:
row.answer = f"{row.answer or ''}{answer_delta}"
db.session.commit()
self._queue_manager.publish(
@@ -562,12 +448,7 @@ def _event_index(data: dict[str, Any]) -> int:
def _string_or_none(value: Any) -> str | None:
if not isinstance(value, str):
return None
normalized = value.strip()
if not normalized or normalized.lower() in {"none", "null"}:
return None
return normalized
return value if isinstance(value, str) and value else None
def _json_or_text(value: Any) -> str | None:
@@ -587,18 +468,6 @@ def _tool_labels(tool: str | None) -> str:
return json.dumps({tool: {"en_US": tool, "zh_Hans": tool}}, ensure_ascii=False)
def _suffix_prefix_overlap_length(text: str, prefix_source: str) -> int:
max_length = min(len(text), len(prefix_source))
for length in range(max_length, 0, -1):
if text.endswith(prefix_source[:length]):
return length
return 0
def _is_empty_answer_only_thought(row: MessageAgentThought) -> bool:
return not any((row.thought, row.answer, row.tool, row.tool_input, row.observation))
class AgentAppRunner:
"""Runs one Agent App conversation turn against the Agent backend."""
@@ -631,9 +500,7 @@ class AgentAppRunner:
model_name: str,
queue_manager: AppQueueManager,
session_scope_snapshot_id: str | None | _DefaultSessionScopeSnapshotId = _DEFAULT_SESSION_SCOPE_SNAPSHOT_ID,
agent_runtime_exit_intent: AgentRuntimeExitIntent = "suspend",
) -> None:
preserve_session = agent_runtime_exit_intent == "suspend"
scope = self._build_session_scope(
dify_context=dify_context,
agent_id=agent_id,
@@ -655,11 +522,10 @@ class AgentAppRunner:
idempotency_key=message_id,
stored=stored,
message_id=message_id,
suspend_on_exit=preserve_session,
)
create_response = self._agent_backend_client.create_run(runtime.request)
terminal, process_recorder = self._consume_stream(
terminal, streamed_answer = self._consume_stream(
create_response.run_id,
dify_context=dify_context,
message_id=message_id,
@@ -669,9 +535,6 @@ class AgentAppRunner:
)
if isinstance(terminal, AgentBackendDeferredToolCallInternalEvent):
if not preserve_session:
self._mark_session_cleaned(scope=scope, backend_run_id=terminal.run_id)
raise AgentBackendError("Agent App finalization cannot pause for human input.")
# ENG-635: the agent asked a human. End this turn with the question and
# a conversation-owned HITL form; a form submission resumes the run.
self._pause_for_ask_human(
@@ -689,53 +552,73 @@ class AgentAppRunner:
return
if not isinstance(terminal, AgentBackendRunSucceededInternalEvent):
if isinstance(terminal, AgentBackendRunFailedInternalEvent):
raise _agent_backend_failure_to_exception(terminal)
raise AgentBackendError("Agent backend run did not complete successfully.")
error = getattr(terminal, "error", None) or "Agent backend run did not complete successfully."
raise AgentBackendError(str(error))
answer = self._terminal_output_to_answer(terminal.output)
try:
process_recorder.trim_answer_suffix(answer)
except Exception:
db.session.rollback()
logger.warning(
"Failed to trim Agent App answer text: run_id=%s message_id=%s",
terminal.run_id,
message_id,
exc_info=True,
)
if preserve_session:
superseded_sessions = self._load_superseded_sessions(scope=scope)
self._publish_terminal_answer(
queue_manager=queue_manager,
model_name=model_name,
answer=answer,
query=query,
usage=_llm_usage_from_agent_backend(terminal.usage),
)
session_saved = self._save_session(
scope=scope,
backend_run_id=terminal.run_id,
snapshot=terminal.session_snapshot,
runtime_layer_specs=extract_runtime_layer_specs(runtime.request.composition),
)
if session_saved:
self._cleanup_superseded_sessions(superseded_sessions)
else:
# The backend has already accepted a terminal success with
# delete-on-exit semantics. Local publish/persistence errors must
# not keep the API-side session row active, and cleanup failures
# must not replace the original publish/error outcome.
try:
self._publish_terminal_answer(
queue_manager=queue_manager,
model_name=model_name,
answer=answer,
query=query,
usage=_llm_usage_from_agent_backend(terminal.usage),
)
finally:
self._mark_session_cleaned(scope=scope, backend_run_id=terminal.run_id)
answer = self._extract_answer(terminal.output)
self._publish_terminal_answer(
queue_manager=queue_manager,
model_name=model_name,
answer=answer,
query=query,
streamed_answer=streamed_answer,
usage=_llm_usage_from_agent_backend(terminal.usage),
)
self._save_session(
scope=scope,
backend_run_id=terminal.run_id,
snapshot=terminal.session_snapshot,
runtime_layer_specs=extract_runtime_layer_specs(runtime.request.composition),
)
def run_stateless(
self,
*,
dify_context: DifyRunContext,
agent_id: str,
agent_config_snapshot_id: str,
agent_config_version_kind: Literal["snapshot", "draft", "build_draft"] = "snapshot",
agent_soul: AgentSoulConfig,
conversation_id: str,
query: str,
idempotency_key: str,
session_scope_snapshot_id: str | None | _DefaultSessionScopeSnapshotId = _DEFAULT_SESSION_SCOPE_SNAPSHOT_ID,
) -> None:
"""Run the Agent backend without creating Dify chat message records.
This path is used by build-chat finalization: the API must trigger the
backend side effects in the existing conversation session, but it must
not persist a synthetic user/assistant turn, update API-side runtime
session rows, or set up HITL state that depends on one.
"""
scope = self._build_session_scope(
dify_context=dify_context,
agent_id=agent_id,
agent_config_snapshot_id=agent_config_snapshot_id,
conversation_id=conversation_id,
session_scope_snapshot_id=session_scope_snapshot_id,
)
runtime = self._build_runtime(
dify_context=dify_context,
agent_id=agent_id,
agent_config_snapshot_id=agent_config_snapshot_id,
agent_config_version_kind=agent_config_version_kind,
agent_soul=agent_soul,
conversation_id=conversation_id,
query=query,
idempotency_key=idempotency_key,
stored=self._session_store.load_active_session(scope),
message_id=None,
)
create_response = self._agent_backend_client.create_run(runtime.request)
status = self._agent_backend_client.wait_run(
create_response.run_id,
timeout_seconds=dify_config.APP_MAX_EXECUTION_TIME,
)
if status.status != "succeeded":
error = getattr(status, "error", None) or f"Agent backend run ended with status {status.status}."
raise AgentBackendError(str(error))
def _build_session_scope(
self,
@@ -771,7 +654,6 @@ class AgentAppRunner:
idempotency_key: str,
stored: StoredAgentAppSession | None,
message_id: str | None,
suspend_on_exit: bool,
) -> AgentAppRuntimeRequest:
session_snapshot = stored.session_snapshot if stored is not None else None
deferred_tool_results = (
@@ -791,7 +673,6 @@ class AgentAppRunner:
idempotency_key=idempotency_key,
session_snapshot=session_snapshot,
deferred_tool_results=deferred_tool_results,
suspend_on_exit=suspend_on_exit,
)
)
@@ -894,61 +775,46 @@ class AgentAppRunner:
model_name: str,
query: str | None,
):
"""Consume backend events while preserving raw recorder granularity."""
"""Consume backend events while preserving raw recorder granularity.
Process events are recorded immediately for observability. Only the
final assistant text deltas sent through the EasyUI queue are debounced,
with flushes happening on later stream events or terminal boundaries.
"""
terminal = None
streamed_answer_parts: list[str] = []
text_delta_debouncer = _TextDeltaDebouncer(debounce_seconds=self._text_delta_debounce_seconds)
process_recorder = _AgentProcessRecorder(
dify_context=dify_context,
message_id=message_id,
queue_manager=queue_manager,
)
text_delta_debouncer = _TextDeltaDebouncer(debounce_seconds=self._text_delta_debounce_seconds)
def persist_answer_text(content_delta: str) -> None:
try:
process_recorder.append_answer_text(content_delta)
except Exception:
db.session.rollback()
logger.warning(
"Failed to persist Agent App answer text: run_id=%s message_id=%s",
run_id,
message_id,
exc_info=True,
)
publish_agent_message_delta(
queue_manager=queue_manager,
model_name=model_name,
delta=content_delta,
user_query=query,
)
def flush_pending_agent_message_text() -> None:
def flush_pending_text() -> None:
pending_text = text_delta_debouncer.flush()
if pending_text:
persist_answer_text(pending_text)
publish_text_delta(
queue_manager=queue_manager,
model_name=model_name,
delta=pending_text,
user_query=query,
)
for public_event in self._agent_backend_client.stream_events(run_id):
if queue_manager.is_stopped():
flush_pending_agent_message_text()
flush_pending_text()
self._cancel_run(run_id)
raise GenerateTaskStoppedError()
for internal_event in self._event_adapter.adapt(public_event):
if queue_manager.is_stopped():
flush_pending_agent_message_text()
flush_pending_text()
self._cancel_run(run_id)
raise GenerateTaskStoppedError()
if internal_event.type in (
AgentBackendInternalEventType.RUN_STARTED,
AgentBackendInternalEventType.STREAM_EVENT,
AgentBackendInternalEventType.AGENT_MESSAGE_DELTA,
):
if isinstance(internal_event, AgentBackendAgentMessageDeltaInternalEvent):
debounced_delta = text_delta_debouncer.push(internal_event.delta)
if debounced_delta:
persist_answer_text(debounced_delta)
continue
if isinstance(internal_event, AgentBackendStreamInternalEvent):
flush_pending_agent_message_text()
try:
process_recorder.handle_stream_event(internal_event)
except Exception:
@@ -960,15 +826,26 @@ class AgentAppRunner:
internal_event.event_kind,
exc_info=True,
)
text_delta = self._extract_stream_text_delta(internal_event)
if text_delta:
streamed_answer_parts.append(text_delta)
debounced_delta = text_delta_debouncer.push(text_delta)
if debounced_delta:
publish_text_delta(
queue_manager=queue_manager,
model_name=model_name,
delta=debounced_delta,
user_query=query,
)
continue
continue
flush_pending_agent_message_text()
flush_pending_text()
terminal = internal_event
break
if terminal is not None:
break
flush_pending_agent_message_text()
return terminal, process_recorder
flush_pending_text()
return terminal, "".join(streamed_answer_parts)
def _cancel_run(self, run_id: str) -> None:
try:
@@ -990,15 +867,42 @@ class AgentAppRunner:
model_name: str,
answer: str,
query: str | None,
streamed_answer: str,
usage: LLMUsage | None,
) -> None:
"""Finish a successful turn from the backend terminal output."""
publish_text_answer(
"""Finish a successful streamed turn without duplicating the final text."""
if not answer and streamed_answer:
answer = streamed_answer
if not streamed_answer:
publish_text_answer(
queue_manager=queue_manager,
model_name=model_name,
answer=answer,
user_query=query,
usage=usage,
)
return
if answer.startswith(streamed_answer):
publish_text_delta(
queue_manager=queue_manager,
model_name=model_name,
delta=answer[len(streamed_answer) :],
user_query=query,
)
elif answer != streamed_answer:
logger.warning(
"Agent App streamed answer does not match terminal output; "
"using terminal output for message persistence."
)
publish_message_end(
queue_manager=queue_manager,
model_name=model_name,
answer=answer,
usage=usage,
user_query=query,
usage=usage,
)
def _save_session(
@@ -1010,7 +914,7 @@ class AgentAppRunner:
runtime_layer_specs: Any,
pending_form_id: str | None = None,
pending_tool_call_id: str | None = None,
) -> bool:
) -> None:
try:
self._session_store.save_active_snapshot(
scope=scope,
@@ -1020,7 +924,6 @@ class AgentAppRunner:
pending_form_id=pending_form_id,
pending_tool_call_id=pending_tool_call_id,
)
return True
except Exception:
logger.warning(
"Failed to persist Agent App conversation session snapshot: "
@@ -1031,91 +934,9 @@ class AgentAppRunner:
scope.agent_id,
exc_info=True,
)
return False
def _load_superseded_sessions(self, *, scope: AgentAppSessionScope) -> list[StoredAgentAppSession]:
try:
stored_sessions = self._session_store.list_active_sessions_for_conversation(
tenant_id=scope.tenant_id,
app_id=scope.app_id,
conversation_id=scope.conversation_id,
)
except Exception:
logger.warning(
"Failed to load existing Agent App conversation sessions before snapshot save: "
"tenant_id=%s app_id=%s conversation_id=%s agent_id=%s",
scope.tenant_id,
scope.app_id,
scope.conversation_id,
scope.agent_id,
exc_info=True,
)
return []
return [stored for stored in stored_sessions if stored.scope != scope]
def _cleanup_superseded_sessions(self, stored_sessions: list[StoredAgentAppSession]) -> None:
for stored_session in stored_sessions:
try:
if stored_session.runtime_layer_specs:
payload = AgentBackendSessionCleanupPayload(
session_snapshot=stored_session.session_snapshot,
runtime_layer_specs=stored_session.runtime_layer_specs,
idempotency_key=(
f"{stored_session.scope.tenant_id}:{stored_session.scope.app_id}:"
f"{stored_session.scope.conversation_id}:{stored_session.scope.agent_id}:"
f"{stored_session.scope.agent_config_snapshot_id or 'no-config'}:"
f"superseded-session-cleanup:{stored_session.backend_run_id or 'no-run'}"
),
metadata={
"tenant_id": stored_session.scope.tenant_id,
"app_id": stored_session.scope.app_id,
"conversation_id": stored_session.scope.conversation_id,
"agent_id": stored_session.scope.agent_id,
"agent_config_snapshot_id": stored_session.scope.agent_config_snapshot_id,
"previous_agent_backend_run_id": stored_session.backend_run_id,
},
)
cleanup_conversation_agent_runtime_session.delay(payload.model_dump(mode="json"))
except Exception:
logger.warning(
"Failed to enqueue Agent backend cleanup for superseded Agent App session: "
"tenant_id=%s app_id=%s conversation_id=%s agent_id=%s backend_run_id=%s",
stored_session.scope.tenant_id,
stored_session.scope.app_id,
stored_session.scope.conversation_id,
stored_session.scope.agent_id,
stored_session.backend_run_id,
exc_info=True,
)
def _mark_session_cleaned(
self,
*,
scope: AgentAppSessionScope,
backend_run_id: str,
) -> None:
"""Best-effort delete-on-exit cleanup for the API-side session row.
Once the Agent backend reaches a terminal event, cleanup persistence
must not replace the original publish/error outcome for that turn.
"""
try:
self._session_store.mark_cleaned(scope=scope, backend_run_id=backend_run_id)
except Exception:
logger.warning(
"Failed to retire Agent App conversation session after delete-on-exit: "
"tenant_id=%s app_id=%s conversation_id=%s agent_id=%s backend_run_id=%s",
scope.tenant_id,
scope.app_id,
scope.conversation_id,
scope.agent_id,
backend_run_id,
exc_info=True,
)
@staticmethod
def _terminal_output_to_answer(output: JsonValue) -> str:
def _extract_answer(output: JsonValue) -> str:
"""Normalize the backend's terminal output to assistant text.
Free-text Agent Apps return a plain string; if a structured output is
@@ -1133,5 +954,27 @@ class AgentAppRunner:
return json.dumps(output, ensure_ascii=False)
return json.dumps(output, ensure_ascii=False)
@staticmethod
def _extract_stream_text_delta(event: AgentBackendStreamInternalEvent) -> str | None:
data = event.data
if not isinstance(data, dict):
return None
if data.get("event_kind") == "part_delta":
delta = data.get("delta")
if isinstance(delta, dict) and delta.get("part_delta_kind") == "text":
content_delta = delta.get("content_delta")
if isinstance(content_delta, str):
return content_delta
if data.get("event_kind") == "part_start":
part = data.get("part")
if isinstance(part, dict) and part.get("part_kind") == "text":
content = part.get("content")
if isinstance(content, str):
return content
return None
__all__ = ["AgentAppRunner", "publish_message_end", "publish_text_answer", "publish_text_delta"]
@@ -74,7 +74,6 @@ class AgentAppRuntimeBuildContext:
session_snapshot: CompositorSessionSnapshot | None = None
# ENG-638: set when resuming a chat turn after a submitted ask_human form.
deferred_tool_results: DeferredToolResultsPayload | None = None
suspend_on_exit: bool = True
@dataclass(frozen=True, slots=True)
@@ -164,7 +163,6 @@ class AgentAppRuntimeRequestBuilder:
# no frontend-internal {{#…#}} marker ever reaches the model.
agent_soul_prompt=expand_prompt_mentions(agent_soul.prompt.system_prompt, soul_prompt_resolver).strip()
or None,
agent_config_version_kind=context.agent_config_version_kind,
user_prompt=context.user_query,
tools=tool_layers.plugin_tools,
core_tools=tool_layers.core_tools,
@@ -175,7 +173,6 @@ class AgentAppRuntimeRequestBuilder:
shell_config=build_shell_layer_config(agent_soul),
session_snapshot=context.session_snapshot,
deferred_tool_results=context.deferred_tool_results,
suspend_on_exit=context.suspend_on_exit,
idempotency_key=context.idempotency_key,
metadata=metadata,
)

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