Compare commits

..
Author SHA1 Message Date
yyh 8be4adb97b ci: run external e2e in main ci 2026-07-10 05:07:00 +08:00
yyhandGitHub eb74946b25 fix(web): save agent configure draft on route leave (#38620) 2026-07-09 13:55:34 +00:00
QuantumGhostandGitHub b5e35fc2fc fix: remove hardcoded sandbox path in configuration file (#38618) 2026-07-09 13:30:56 +00:00
wangxiaoleiandGitHub 7bfbb2bbe8 fix: fix auth prefix duplicate (#38616) 2026-07-09 12:55:58 +00:00
QuantumGhostandGitHub d177998255 chore: Bump version to 1.16.0-rc1 (#38600) 2026-07-09 11:42:09 +00:00
wangxiaoleiGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
74ee665af6 fix: fix miss session param (#38612)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-09 11:26:16 +00:00
Stephen ZhouandGitHub d655153a3d fix(web): update docs links (#38591) 2026-07-09 11:25:31 +00:00
zyssyz123GitHub盐粒 YanliyyhJoelautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>QuantumGhostyyh盐粒 Yanli
925f97be20 feat: daily sync (#38593)
Co-authored-by: 盐粒 Yanli <mail@yanli.one>
Co-authored-by: yyh <yuanyouhuilyz@gmail.com>
Co-authored-by: Joel <iamjoel007@gmail.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: QuantumGhost <obelisk.reg+git@gmail.com>
Co-authored-by: yyh <92089059+lyzno1@users.noreply.github.com>
Co-authored-by: 盐粒 Yanli <yanli@dify.ai>
2026-07-09 10:35:15 +00:00
9252d81826 docs: remove Dify Premium on AWS Marketplace section from all READMEs (#38607)
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-07-09 10:09:59 +00:00
Harsh KashyapGitHubHarsh Kashyapautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Harsh Kashyap
9d5819a9c1 fix(api): ignore invalid utf8 cache payloads (#37835)
Co-authored-by: Harsh Kashyap <Harsh23Kashyap@users.noreply.github.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Harsh Kashyap <harshkashyap@Harshs-MacBook-Pro.local>
2026-07-09 09:52:49 +00:00
Ingram ZGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
99e3b1a401 fix: harden workflow archive DB retries (#38170)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-09 09:46:28 +00:00
Asuka MinatoGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
45957225cd chore: batch example #38419 (#38474)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-09 09:14:52 +00:00
2f99652203 fix: chunk workflow failure tracking data (#38598)
Co-authored-by: CodingOnStar <hanxujiang@dify.com>
2026-07-09 08:33:47 +00:00
dc1131b6df refactor(tests): replace logger mocks with caplog in trace provider tests (#38569)
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-09 08:33:28 +00:00
Stephen ZhouandGitHub a5a7c762a3 refactor(web): split app context state atoms (#38588) 2026-07-09 07:42:02 +00:00
Coding On StarGitHubCodingOnStarautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
063e390c5d fix(web): preserve attribution from auth redirect (#38583)
Co-authored-by: CodingOnStar <hanxujiang@dify.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-09 07:40:55 +00:00
JoelandGitHub 3b3c25273a fix: guard chat tree against out-of-order parents (#38590) 2026-07-09 07:05:05 +00:00
Xiyuan ChenandGitHub 3775f569f1 fix: run user_connect authorization inside Flask app context (#38585) 2026-07-09 06:00:29 +00:00
chaririGitHubAsuka Minatoautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
6540d178c6 refactor(api): migrate service app endpoints to BaseModel (#37960)
Co-authored-by: Asuka Minato <i@asukaminato.eu.org>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-09 04:23:47 +00:00
chaririandGitHub 458aa4892d refactor(api): migrate web chat endpoints to BaseModel (#37962) 2026-07-09 04:23:13 +00:00
chaririGitHubAsuka Minatoautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
89d5f74a40 refactor(api): migrate dataset endpoints to BaseModel (#37957)
Co-authored-by: Asuka Minato <i@asukaminato.eu.org>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-09 04:22:30 +00:00
9bb3b1fa98 refactor(api): migrate snippet workspace endpoints to BaseModel (#37956)
Co-authored-by: Asuka Minato <i@asukaminato.eu.org>
2026-07-09 04:21:27 +00:00
chaririGitHubAsuka Minatoautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>Byron Wang
3cd8d850fa refactor(api): migrate workspace model endpoints to BaseModel (#37963)
Co-authored-by: Asuka Minato <i@asukaminato.eu.org>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Byron Wang <byron@dify.ai>
2026-07-09 04:05:32 +00:00
Stephen ZhouandGitHub c5cef80ea4 chore: generate fastopenapi console contracts (#38580) 2026-07-09 03:52:42 +00:00
Stephen ZhouandGitHub cee88de684 refactor(web): add prefetched query atom (#38572) 2026-07-09 02:47:51 +00:00
非法操作andGitHub 5741f8f9d1 fix: when delete custom model remove its cache (#38577) 2026-07-09 02:12:23 +00:00
Stephen ZhouGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
512f39dede chore: upgrade TypeScript 7 (#38575)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-09 01:42:40 +00:00
Xiyuan ChenGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
d72ee32ba1 fix: preserve ResponseStreamFilter state across workflow pause/resume (#38540)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-09 01:05:47 +00:00
yyhandGitHub 2b35f48d77 test(e2e): fix agent build note runtime connection (#38574) 2026-07-09 00:36:20 +00:00
Stephen ZhouandGitHub caf1a22020 refactor(web): sync app context effects with jotai (#38570) 2026-07-08 15:55:25 +00:00
yyhandGitHub 9416d04163 test(e2e): stabilize agent build draft note scenario (#38571) 2026-07-08 15:38:47 +00:00
Stephen ZhouandGitHub 23b936aeb5 refactor(web): remove app context provider (#38568) 2026-07-08 13:52:45 +00:00
eb3362e646 perf: batch-load messages in advanced-chat workflow run list to remove N+1 (#38359)
Co-authored-by: Yunlu Wen <yunlu.wen@dify.ai>
2026-07-08 13:32:32 +00:00
+3 98d9b11f7b chore(agent-v2): sync changes (#38513)
Co-authored-by: Joel <iamjoel007@gmail.com>
Co-authored-by: 林玮 (Jade Lin) <linw1995@icloud.com>
Co-authored-by: 盐粒 Yanli <mail@yanli.one>
Co-authored-by: Asuka Minato <i@asukaminato.eu.org>
Co-authored-by: Jashwanth Reddy Gummula <gmrnlg1971@gmail.com>
Co-authored-by: WH-2099 <wh2099@pm.me>
Co-authored-by: 非法操作 <hjlarry@163.com>
Co-authored-by: wangxiaolei <fatelei@gmail.com>
Co-authored-by: FFXN <31929997+FFXN@users.noreply.github.com>
Co-authored-by: Yansong Zhang <916125788@qq.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-08 12:05:04 +00:00
Stephen ZhouGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
d67123e5fd refactor(web): remove remaining app context consumers (#38565)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-08 11:28:25 +00:00
yyhandGitHub 1340bef096 test: scope agent build draft e2e selector (#38563) 2026-07-08 10:32:57 +00:00
Stephen ZhouGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
94ba597d32 refactor(web): migrate shell navigation app context consumers (#38560)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-08 10:11:27 +00:00
chaririandGitHub 9465dc2477 refactor(api): remove remaining legacy field remnants (#37967) 2026-07-08 09:59:10 +00:00
chaririandGitHub 74f177efe6 refactor(api): migrate workspace tool endpoints to BaseModel (#37959) 2026-07-08 09:56:04 +00:00
Stephen ZhouandGitHub af5b08ac5c refactor(web): migrate agent v2 app context consumers (#38558) 2026-07-08 09:42:49 +00:00
Stephen ZhouandGitHub eca2d419b2 refactor(web): migrate workflow app context consumers (#38552) 2026-07-08 09:01:30 +00:00
Asuka MinatoGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
a278d21741 test: more caplog (#38452)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-08 08:24:31 +00:00
Stephen ZhouGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
dce45ef6ae refactor(web): migrate account settings app context consumers (#38544)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-08 08:15:54 +00:00
CrazywoolaGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
5ffd4345dc fix: display errors for oauth page (#38546)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-08 08:07:14 +00:00
yyhandGitHub 16ae1dd131 chore: remove superpowers artifacts (#38547) 2026-07-08 07:59:32 +00:00
Xiyuan ChenandGitHub ae1e180b54 fix(cli): --insecure also skips TLS certificate verification (#38531) 2026-07-08 07:58:31 +00:00
zyssyz123GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
5d61318860 feat(api): use billing quota for credit pool (#38028)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-08 07:10:18 +00:00
Stephen ZhouandGitHub 5a210cf03e refactor(web): migrate billing app context consumers (#38541) 2026-07-08 07:02:47 +00:00
YuziandGitHub 3523da508f fix(web): guard invite-settings activate button against double-click (#38337) 2026-07-08 06:28:43 +00:00
Stephen ZhouGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
503d80be1d refactor(web): migrate plugins and tools app context consumers (#38533)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-08 06:26:56 +00:00
EvanGitHubQuantumGhostautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
98f67e9c82 fix(ci): make no-new-getattr guard stable in shallow PR checkouts (#38480)
Co-authored-by: QuantumGhost <obelisk.reg+git@gmail.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-08 06:22:59 +00:00
i晟GitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
af4538e942 fix: raise clear error on unsupported language in execute_code (#38448)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-08 06:20:10 +00:00
yyhandGitHub 10cf9be3c7 test(web): align auth e2e with console home (#38538) 2026-07-08 06:17:14 +00:00
yyhandGitHub 41fb479957 fix(web): cache server console context (#38535) 2026-07-08 06:16:59 +00:00
yyhandGitHub 6ac7ff6586 chore(deps): upgrade vite-plus toolchain (#38534) 2026-07-08 04:33:10 +00:00
Stephen ZhouandGitHub 76a6cd3335 refactor(web): migrate app context consumers (#38530) 2026-07-08 03:51:50 +00:00
yyhandGitHub 6c0aa3ed0d test(dify-ui): remove brittle primitive assertions (#38529) 2026-07-08 03:24:22 +00:00
68d8328b9c chore: clean Db session from service (#38227)
Co-authored-by: chariri <w@chariri.moe>
Co-authored-by: WH-2099 <wh2099@pm.me>
2026-07-08 03:07:27 +00:00
Stephen ZhouGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
ee0068eed4 refactor(web): migrate dataset access context (#38523)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-08 02:35:54 +00:00
L1nSn0wGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
915655683c refactor(openapi): resource-oriented paths for /openapi/v1 + difyctl version gate (#38367)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-08 02:22:59 +00:00
1189 changed files with 40816 additions and 15028 deletions
+9
View File
@@ -53,6 +53,10 @@ 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'
@@ -95,6 +99,7 @@ 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:
@@ -322,6 +327,8 @@ 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:
@@ -380,6 +387,8 @@ 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
+6 -9
View File
@@ -2,6 +2,10 @@ name: Style check
on:
workflow_call:
inputs:
base-rev:
required: true
type: string
concurrency:
group: style-${{ github.head_ref || github.run_id }}
@@ -33,6 +37,7 @@ 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'
@@ -54,17 +59,9 @@ 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 --mode ci --merge-target main
run: uv run --project api python scripts/check_no_new_getattr.py --base-rev "${{ inputs.base-rev }}"
- name: Run Type Checks
if: steps.changed-files.outputs.any_changed == 'true'
+10 -1
View File
@@ -96,6 +96,15 @@ 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
@@ -111,5 +120,5 @@ jobs:
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
with:
name: e2e-logs
path: e2e/.logs
path: e2e/.logs/**
retention-days: 7
@@ -1,30 +0,0 @@
# 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,8 +125,6 @@ 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,6 +663,9 @@ 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,6 +18,7 @@ from clients.agent_backend.errors import (
AgentBackendValidationError,
)
from clients.agent_backend.event_adapter import (
AgentBackendAgentMessageDeltaInternalEvent,
AgentBackendDeferredToolCallInternalEvent,
AgentBackendInternalEvent,
AgentBackendInternalEventType,
@@ -46,6 +47,11 @@ 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",
@@ -57,6 +63,7 @@ __all__ = [
"WORKFLOW_NODE_JOB_PROMPT_LAYER_ID",
"WORKFLOW_USER_PROMPT_LAYER_ID",
"AgentBackendAgentAppRunInput",
"AgentBackendAgentMessageDeltaInternalEvent",
"AgentBackendDeferredToolCallInternalEvent",
"AgentBackendError",
"AgentBackendHTTPError",
@@ -73,6 +80,8 @@ __all__ = [
"AgentBackendRunRequestBuilder",
"AgentBackendRunStartedInternalEvent",
"AgentBackendRunSucceededInternalEvent",
"AgentBackendSessionCleanupPayload",
"AgentBackendSessionCleanupResult",
"AgentBackendStreamError",
"AgentBackendStreamInternalEvent",
"AgentBackendTransportError",
@@ -82,6 +91,7 @@ __all__ = [
"FakeAgentBackendRunClient",
"FakeAgentBackendScenario",
"RuntimeLayerSpec",
"cleanup_agent_backend_session",
"create_agent_backend_run_client",
"extract_runtime_layer_specs",
"redact_for_agent_backend_log",
@@ -5,6 +5,9 @@ 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
@@ -32,6 +35,7 @@ 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"
@@ -61,6 +65,13 @@ 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."""
@@ -99,6 +110,7 @@ class AgentBackendRunCancelledInternalEvent(AgentBackendInternalEventBase):
type AgentBackendInternalEvent = Annotated[
AgentBackendRunStartedInternalEvent
| AgentBackendStreamInternalEvent
| AgentBackendAgentMessageDeltaInternalEvent
| AgentBackendDeferredToolCallInternalEvent
| AgentBackendRunSucceededInternalEvent
| AgentBackendRunFailedInternalEvent
@@ -121,6 +133,14 @@ 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 [
+70 -11
View File
@@ -11,8 +11,9 @@ composition-driven.
from __future__ import annotations
import re
from collections.abc import Mapping
from typing import ClassVar
from typing import ClassVar, Literal
from agenton.compositor import CompositorSessionSnapshot
from agenton.compositor.schemas import LayerSessionSnapshot
@@ -46,7 +47,6 @@ from dify_agent.protocol import (
LayerExitSignals,
RunComposition,
RunLayerSpec,
RunPurpose,
RuntimeLayerSpec,
)
from pydantic import BaseModel, ConfigDict, Field, JsonValue, field_validator
@@ -63,6 +63,7 @@ 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(
@@ -104,6 +105,59 @@ 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."""
@@ -163,7 +217,7 @@ class AgentBackendWorkflowNodeRunInput(BaseModel):
workflow_node_job_prompt: str
user_prompt: str
agent_soul_prompt: str | None = None
purpose: RunPurpose = "workflow_node"
agent_config_version_kind: AgentConfigVersionKind = "snapshot"
idempotency_key: str | None = None
output: AgentBackendOutputConfig | None = None
tools: DifyPluginToolsLayerConfig | None = None
@@ -212,7 +266,7 @@ class AgentBackendAgentAppRunInput(BaseModel):
execution_context: DifyExecutionContextLayerConfig
user_prompt: str
agent_soul_prompt: str | None = None
purpose: RunPurpose = "agent_app"
agent_config_version_kind: AgentConfigVersionKind = "snapshot"
idempotency_key: str | None = None
output: AgentBackendOutputConfig | None = None
tools: DifyPluginToolsLayerConfig | None = None
@@ -261,13 +315,17 @@ class AgentBackendRunRequestBuilder:
prompt.
"""
layers: list[RunLayerSpec] = []
if run_input.agent_soul_prompt:
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:
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=run_input.agent_soul_prompt),
config=PromptLayerConfig(prefix=agent_soul_prompt),
)
)
@@ -419,7 +477,6 @@ 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,
@@ -467,7 +524,6 @@ 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,
@@ -483,13 +539,17 @@ class AgentBackendRunRequestBuilder:
ask_human / structured output.
"""
layers: list[RunLayerSpec] = []
if run_input.agent_soul_prompt:
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:
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=run_input.agent_soul_prompt),
config=PromptLayerConfig(prefix=agent_soul_prompt),
)
)
@@ -649,7 +709,6 @@ 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,
@@ -0,0 +1,100 @@
"""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(db.session(), email.strip())
account = AccountService.get_account_by_email_with_case_fallback(email.strip(), session=db.session())
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(db.session(), email.strip())
account = AccountService.get_account_by_email_with_case_fallback(email.strip(), session=db.session())
if not account:
click.echo(click.style(f"Account not found for email: {email}", fg="red"))
+122 -36
View File
@@ -1,10 +1,12 @@
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
@@ -12,6 +14,7 @@ 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
@@ -35,6 +38,12 @@ 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)
@@ -57,6 +66,7 @@ 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,
@@ -75,7 +85,7 @@ def _get_archive_candidate_tenant_ids_by_prefix(
if start_from is not None:
conditions.append(WorkflowRun.created_at >= start_from)
tenant_ids = db.session.scalars(
tenant_ids = session.scalars(
sa.select(WorkflowRun.tenant_id).where(*conditions).distinct().order_by(WorkflowRun.tenant_id)
).all()
return list(tenant_ids)
@@ -102,8 +112,80 @@ 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,
@@ -122,7 +204,8 @@ def _resolve_archive_tenant_ids_from_plan(
requested_tenant_ids = []
for prefix in tenant_prefixes:
requested_tenant_ids.extend(
_get_archive_candidate_tenant_ids_by_prefix(
_get_archive_candidate_tenant_ids_with_retry(
session_maker,
prefix,
start_from=start_from,
end_before=end_before,
@@ -143,6 +226,21 @@ 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,
@@ -349,10 +447,6 @@ 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,
@@ -364,37 +458,25 @@ 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:
tenant_ids = _get_archive_candidate_tenant_ids_by_prefix(
prefix,
start_from=start_from,
end_before=plan_end_before,
)
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"]
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,
@@ -574,17 +656,18 @@ 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:
except Exception as exc:
logger.exception("Failed to resolve workflow archive tenant plan")
click.echo(click.style("Failed to resolve workflow archive tenant plan.", fg="red"))
return
raise click.ClickException("Failed to resolve workflow archive tenant plan.") from exc
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
@@ -616,7 +699,10 @@ def archive_workflow_runs(
dry_run=dry_run,
delete_after_archive=delete_after_archive,
)
summary = archiver.run()
try:
summary = archiver.run()
finally:
_safe_remove_scoped_session("archive workflow run command")
click.echo(
click.style(
f"Summary: processed={summary.total_runs_processed}, archived={summary.runs_archived}, "
+3 -4
View File
@@ -25,11 +25,10 @@ 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; keep it "
"off until shellctl is deployed, otherwise every agent run that includes the "
"shell layer will fail."
"Requires the agent backend to be wired with a shellctl entrypoint before "
"shell-using Agent runs are executed."
),
default=False,
default=True,
)
AGENT_APP_TEXT_DELTA_DEBOUNCE_SECONDS: NonNegativeFloat = Field(
+4 -1
View File
@@ -363,7 +363,10 @@ 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."
" Falls back to FILES_URL if not specified.",
" 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,
default="",
)
+16
View File
@@ -6,6 +6,17 @@ 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
@@ -15,3 +26,8 @@ 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(),
)
+6 -1
View File
@@ -1,6 +1,6 @@
from __future__ import annotations
from typing import Any
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field, RootModel, computed_field
@@ -52,6 +52,11 @@ 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(session, app=app_model, account_id=current_user_id)
AppService.star_app(app=app_model, account_id=current_user_id, session=session)
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(session, app=app_model, account_id=current_user_id)
AppService.unstar_app(app=app_model, account_id=current_user_id, session=session)
return SimpleResultResponse(result="success").model_dump(mode="json")
+2 -1
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 InvokeFrom
from core.app.entities.app_invoke_entities import AGENT_RUNTIME_EXIT_INTENT_ARG, InvokeFrom
from core.app.features.rate_limiting.rate_limit import RateLimitGenerator
from core.errors.error import (
ModelCurrentlyNotSupportError,
@@ -416,6 +416,7 @@ 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(db.session(), current_tenant_id)
data_source_api_key_bindings = ApiKeyAuthService.get_provider_auth_list(current_tenant_id, session=db.session())
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(db.session(), current_tenant_id, data)
ApiKeyAuthService.create_provider_auth(current_tenant_id, data, session=db.session())
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(db.session(), current_tenant_id, str(binding_id))
ApiKeyAuthService.delete_provider_auth(current_tenant_id, str(binding_id), session=db.session())
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(db.session(), args.email)
account = AccountService.get_account_by_email_with_case_fallback(args.email, session=db.session())
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(db.session(), email)
account = AccountService.get_account_by_email_with_case_fallback(email, session=db.session())
if account:
raise EmailAlreadyInUseError()
@@ -82,7 +82,7 @@ class ForgotPasswordSendEmailApi(Resource):
else:
language = "en-US"
account = AccountService.get_account_by_email_with_case_fallback(db.session(), args.email)
account = AccountService.get_account_by_email_with_case_fallback(args.email, session=db.session())
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(db.session(), email)
account = AccountService.get_account_by_email_with_case_fallback(email, session=db.session())
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(db.session(), user_info.email)
account = AccountService.get_account_by_email_with_case_fallback(user_info.email, session=db.session())
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(db.session(), current_user)
BillingService.is_tenant_owner_or_admin(current_user, session=db.session())
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(db.session(), current_user)
BillingService.is_tenant_owner_or_admin(current_user, session=db.session())
return BillingService.get_invoices(current_user.email, current_tenant_id)
+25 -25
View File
@@ -30,6 +30,7 @@ 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
@@ -268,21 +269,10 @@ class ErrorDocsResponse(DocumentStatusListResponse):
total: int
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 IndexingEstimateResponse(IndexingEstimate):
tokens: int
total_price: float | int
currency: str
class RetrievalSettingResponse(ResponseModel):
@@ -647,7 +637,7 @@ class DatasetApi(Resource):
else:
data["embedding_available"] = True
return data, 200
return dump_response(DatasetDetailWithPartialMembersResponse, data), 200
@console_ns.doc("update_dataset")
@console_ns.doc(description="Update dataset details")
@@ -688,10 +678,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(
session, current_user, dataset, payload.permission, payload.partial_member_list
current_user, dataset, payload.permission, payload.partial_member_list, session=session
)
dataset = DatasetService.update_dataset(session, dataset_id_str, payload_data, current_user)
dataset = DatasetService.update_dataset(dataset_id_str, payload_data, current_user, session=session)
if dataset is None:
raise NotFound("Dataset not found.")
@@ -717,7 +707,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 result_data, 200
return dump_response(DatasetDetailWithPartialMembersResponse, result_data), 200
@setup_required
@login_required
@@ -760,7 +750,7 @@ class DatasetUseCheckApi(Resource):
dataset_id_str = str(dataset_id)
dataset_is_using = DatasetService.dataset_use_check(dataset_id_str, db.session())
return {"is_using": dataset_is_using}, 200
return UsageCheckResponse(is_using=dataset_is_using).model_dump(mode="json"), 200
@console_ns.route("/datasets/<uuid:dataset_id>/queries")
@@ -901,7 +891,17 @@ class DatasetIndexingEstimateApi(Resource):
except Exception as e:
raise IndexingEstimateError(str(e))
return response.model_dump(), 200
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,
)
@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 ApiKeyList.model_validate({"data": keys}, from_attributes=True).model_dump(mode="json")
return dump_response(ApiKeyList, {"data": keys})
@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 ApiKeyItem.model_validate(api_token, from_attributes=True).model_dump(mode="json"), 200
return dump_response(ApiKeyItem, api_token), 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 {"result": "success"}, 200
return SimpleResultResponse(result="success").model_dump(mode="json"), 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 {"api_base_url": normalize_api_base_url(base)}
return ApiBaseUrlResponse(api_base_url=normalize_api_base_url(base)).model_dump(mode="json")
@console_ns.route("/datasets/retrieval-setting")
@@ -10,16 +10,17 @@ from uuid import UUID
import sqlalchemy as sa
from flask import request, send_file
from flask_restx import Resource
from pydantic import BaseModel, Field, RootModel, field_validator
from pydantic import BaseModel, Field, JsonValue, 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 BinaryFileResponse, SimpleResultMessageResponse, SimpleResultResponse, UrlResponse
from controllers.common.fields import 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,
@@ -29,6 +30,7 @@ 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
@@ -49,7 +51,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, SegmentStatus
from models.enums import IndexingStatus, ProcessRuleMode, 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
@@ -148,8 +150,91 @@ class DocumentWithSegmentsListResponse(ResponseModel):
page: int
class OpaqueObjectResponse(RootModel[dict[str, Any]]):
root: dict[str, Any]
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
register_schema_models(
@@ -165,7 +250,6 @@ register_schema_models(
)
register_response_schema_models(
console_ns,
BinaryFileResponse,
SimpleResultMessageResponse,
SimpleResultResponse,
UrlResponse,
@@ -175,7 +259,11 @@ register_response_schema_models(
DocumentWithSegmentsResponse,
DatasetAndDocumentResponse,
DocumentWithSegmentsListResponse,
OpaqueObjectResponse,
IndexingEstimateResponse,
DocumentDetailResponse,
DocumentSummaryStatusResponse,
ProcessRuleResponse,
DocumentPipelineExecutionLogResponse,
)
@@ -225,7 +313,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[OpaqueObjectResponse.__name__])
@console_ns.response(200, "Process rules retrieved successfully", console_ns.models[ProcessRuleResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -264,7 +352,7 @@ class GetProcessRuleApi(Resource):
mode = dataset_process_rule.mode
rules = dataset_process_rule.rules_dict
return {"mode": mode, "rules": rules, "limits": limits}
return dump_response(ProcessRuleResponse, {"mode": mode, "rules": rules, "limits": limits})
@console_ns.route("/datasets/<uuid:dataset_id>/documents")
@@ -491,7 +579,7 @@ class DatasetInitApi(Resource):
@console_ns.doc(description="Initialize dataset with documents")
@console_ns.expect(console_ns.models[KnowledgeConfig.__name__])
@console_ns.response(
201, "Dataset initialized successfully", console_ns.models[DatasetAndDocumentResponse.__name__]
200, "Dataset initialized successfully", console_ns.models[DatasetAndDocumentResponse.__name__]
)
@console_ns.response(400, "Invalid request parameters")
@setup_required
@@ -557,7 +645,7 @@ class DocumentIndexingEstimateApi(DocumentResource):
@console_ns.response(
200,
"Indexing estimate calculated successfully",
console_ns.models[OpaqueObjectResponse.__name__],
console_ns.models[IndexingEstimateResponse.__name__],
)
@console_ns.response(404, "Document not found")
@console_ns.response(400, "Document already finished")
@@ -578,8 +666,6 @@ 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:
@@ -610,7 +696,18 @@ class DocumentIndexingEstimateApi(DocumentResource):
"English",
dataset_id_str,
)
return estimate_response.model_dump(), 200
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,
)
except LLMBadRequestError:
raise ProviderNotInitializeError(
"No Embedding Model available. Please configure a valid provider "
@@ -623,15 +720,24 @@ class DocumentIndexingEstimateApi(DocumentResource):
except Exception as e:
raise IndexingEstimateError(str(e))
return response, 200
return (
IndexingEstimateResponse(
tokens=0,
total_price=0,
currency="USD",
total_segments=0,
preview=[],
).model_dump(mode="json", exclude_none=True),
200,
)
@console_ns.route("/datasets/<uuid:dataset_id>/batch/<string:batch>/indexing-estimate")
class DocumentBatchIndexingEstimateApi(DocumentResource):
@console_ns.response(
200,
"Batch indexing estimate calculated successfully",
console_ns.models[OpaqueObjectResponse.__name__],
"Indexing estimate calculated successfully",
console_ns.models[IndexingEstimateResponse.__name__],
)
@setup_required
@login_required
@@ -643,7 +749,16 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
dataset_id_str = str(dataset_id)
documents = self.get_batch_documents(dataset_id_str, batch, current_user)
if not documents:
return {"tokens": 0, "total_price": 0, "currency": "USD", "total_segments": 0, "preview": []}, 200
return (
IndexingEstimateResponse(
tokens=0,
total_price=0,
currency="USD",
total_segments=0,
preview=[],
).model_dump(mode="json", exclude_none=True),
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 = []
@@ -717,7 +832,17 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
"English",
dataset_id_str,
)
return response.model_dump(), 200
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,
)
except LLMBadRequestError:
raise ProviderNotInitializeError(
"No Embedding Model available. Please configure a valid provider in the Settings -> Model Provider."
@@ -854,7 +979,7 @@ class DocumentApi(DocumentResource):
"metadata": "Metadata inclusion (all/only/without)",
}
)
@console_ns.response(200, "Document retrieved successfully", console_ns.models[OpaqueObjectResponse.__name__])
@console_ns.response(200, "Document retrieved successfully", console_ns.models[DocumentDetailResponse.__name__])
@console_ns.response(404, "Document not found")
@setup_required
@login_required
@@ -871,46 +996,21 @@ 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 = {"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 = {
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(
{
"id": document.id,
"position": document.position,
"data_source_type": document.data_source_type,
@@ -943,8 +1043,9 @@ class DocumentApi(DocumentResource):
"doc_language": document.doc_language,
"need_summary": document.need_summary if document.need_summary is not None else False,
}
return response, 200
)
exclude = metadata_fields if metadata == "without" else None
return response.model_dump(mode="json", exclude=exclude, exclude_unset=True), 200
@setup_required
@login_required
@@ -990,7 +1091,9 @@ 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 {"url": DocumentService.get_document_download_url(document, db.session())}
return UrlResponse(url=DocumentService.get_document_download_url(document, db.session())).model_dump(
mode="json"
)
@console_ns.route("/datasets/<uuid:dataset_id>/documents/download-zip")
@@ -999,7 +1102,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 generated successfully", console_ns.models[BinaryFileResponse.__name__])
@console_ns.response(200, "ZIP archive downloaded successfully")
@setup_required
@login_required
@account_initialization_required
@@ -1034,6 +1137,7 @@ class DocumentBatchDownloadZipApi(DocumentResource):
)
cleanup = stack.pop_all()
response.call_on_close(cleanup.close)
# response-contract:ignore binary ZIP download response
return response
@@ -1093,7 +1197,7 @@ class DocumentProcessingApi(DocumentResource):
document.is_paused = False
db.session.commit()
return {"result": "success"}, 200
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/metadata")
@@ -1152,7 +1256,9 @@ class DocumentMetadataApi(DocumentResource):
document.updated_at = naive_utc_now()
db.session.commit()
return {"result": "success", "message": "Document metadata updated."}, 200
return SimpleResultMessageResponse(result="success", message="Document metadata updated.").model_dump(
mode="json"
), 200
@console_ns.route("/datasets/<uuid:dataset_id>/documents/status/<string:action>/batch")
@@ -1194,7 +1300,7 @@ class DocumentStatusApi(DocumentResource):
except NotFound as e:
raise NotFound(str(e))
return {"result": "success"}, 200
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/processing/pause")
@@ -1321,7 +1427,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(dataset_id, db.session())
dataset = DatasetService.get_dataset(str(dataset_id), db.session())
if not dataset:
raise NotFound("Dataset not found.")
DatasetService.check_dataset_operator_permission(current_user, dataset, session=db.session())
@@ -1363,15 +1469,15 @@ class WebsiteDocumentSyncApi(DocumentResource):
# sync document
DocumentService.sync_website_document(dataset_id_str, document, db.session())
return {"result": "success"}, 200
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/pipeline-execution-log")
class DocumentPipelineExecutionLogApi(DocumentResource):
@console_ns.response(
200,
"Document pipeline execution log retrieved successfully",
console_ns.models[OpaqueObjectResponse.__name__],
"Pipeline execution log retrieved successfully",
console_ns.models[DocumentPipelineExecutionLogResponse.__name__],
)
@setup_required
@login_required
@@ -1394,18 +1500,16 @@ class DocumentPipelineExecutionLogApi(DocumentResource):
.limit(1)
)
if not log:
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
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
@console_ns.route("/datasets/<uuid:dataset_id>/documents/generate-summary")
@@ -1508,7 +1612,7 @@ class DocumentGenerateSummaryApi(Resource):
dataset_id_str,
)
return {"result": "success"}, 200
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/summary-status")
@@ -1516,7 +1620,11 @@ 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[OpaqueObjectResponse.__name__])
@console_ns.response(
200,
"Summary status retrieved successfully",
console_ns.models[DocumentSummaryStatusResponse.__name__],
)
@console_ns.response(404, "Document not found")
@setup_required
@login_required
@@ -1534,6 +1642,7 @@ 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)
@@ -1559,4 +1668,4 @@ class DocumentSummaryStatusApi(DocumentResource):
session=db.session(),
)
return result, 200
return dump_response(DocumentSummaryStatusResponse, 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 dump_response(SimpleResultResponse, {"result": "success"}), 200
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segment")
+89 -107
View File
@@ -1,20 +1,16 @@
from datetime import datetime
from typing import Any
from uuid import UUID
from flask import request
from flask_restx import Resource, fields, marshal
from pydantic import BaseModel, Field, RootModel
from flask_restx import Resource
from pydantic import AliasChoices, BaseModel, Field, field_validator
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 (
get_or_create_model,
query_params_from_model,
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 import console_ns
from controllers.console.app.wraps import with_session
from controllers.console.datasets.error import DatasetNameDuplicateError
@@ -28,21 +24,9 @@ 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 (
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 fields.dataset_fields import DatasetDetailResponse
from libs.helper import dump_response
from libs.login import login_required
from models import Account
from services.dataset_service import DatasetService
@@ -51,50 +35,10 @@ 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, object]
settings: dict[str, Any]
class ExternalDatasetCreatePayload(BaseModel):
@@ -102,15 +46,13 @@ 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, object] | None = Field(default=None)
external_retrieval_model: dict[str, Any] | None = None
class ExternalHitTestingPayload(BaseModel):
query: str
external_retrieval_model: dict[str, object] | None = Field(default=None)
metadata_filtering_conditions: dict[str, object] | None = Field(
default=None,
)
external_retrieval_model: dict[str, Any] | None = None
metadata_filtering_conditions: dict[str, Any] | None = None
class BedrockRetrievalPayload(BaseModel):
@@ -125,7 +67,7 @@ class ExternalApiTemplateListQuery(BaseModel):
keyword: str | None = Field(default=None, description="Search keyword")
class ExternalKnowledgeDatasetBindingResponse(ResponseModel):
class ExternalKnowledgeApiBindingResponse(ResponseModel):
id: str
name: str
@@ -135,22 +77,52 @@ class ExternalKnowledgeApiResponse(ResponseModel):
tenant_id: str
name: str
description: str
settings: dict[str, Any] | None = Field(default=None)
dataset_bindings: list[ExternalKnowledgeDatasetBindingResponse] = Field(default_factory=list)
settings: dict[str, Any] | None = Field(validation_alias=AliasChoices("settings_dict", "settings"))
dataset_bindings: list[ExternalKnowledgeApiBindingResponse]
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
total: int | None
page: int
class ExternalRetrievalTestResponse(RootModel[dict[str, Any] | list[dict[str, Any]]]):
root: dict[str, Any] | list[dict[str, Any]]
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]
register_schema_models(
@@ -163,9 +135,16 @@ register_schema_models(
)
register_response_schema_models(
console_ns,
UsageCountResponse,
DatasetDetailResponse,
ExternalKnowledgeApiBindingResponse,
ExternalKnowledgeApiResponse,
ExternalKnowledgeApiListResponse,
ExternalRetrievalTestResponse,
ExternalHitTestingQueryResponse,
ExternalHitTestingRecordResponse,
ExternalHitTestingResponse,
BedrockRetrievalRecordResponse,
BedrockRetrievalResponse,
)
@@ -189,24 +168,26 @@ class ExternalApiTemplateListApi(Resource):
external_knowledge_apis, total = ExternalDatasetService.get_external_knowledge_apis(
query.page, query.limit, current_tenant_id, query.keyword
)
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
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
@setup_required
@login_required
@account_initialization_required
@console_ns.doc("create_external_api_template")
@console_ns.doc(description="Create external knowledge API template")
@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
@@ -229,7 +210,7 @@ class ExternalApiTemplateListApi(Resource):
except services.errors.dataset.DatasetNameDuplicateError:
raise DatasetNameDuplicateError()
return external_knowledge_api.to_dict(), 201
return dump_response(ExternalKnowledgeApiResponse, external_knowledge_api), 201
@console_ns.route("/datasets/external-knowledge-api/<uuid:external_knowledge_api_id>")
@@ -256,17 +237,21 @@ class ExternalApiTemplateApi(Resource):
if external_knowledge_api is None:
raise NotFound("API template not found.")
return external_knowledge_api.to_dict(), 200
return dump_response(ExternalKnowledgeApiResponse, external_knowledge_api), 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
@@ -284,7 +269,7 @@ class ExternalApiTemplateApi(Resource):
session=session,
)
return external_knowledge_api.to_dict(), 200
return dump_response(ExternalKnowledgeApiResponse, external_knowledge_api), 200
@setup_required
@login_required
@@ -322,7 +307,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 {"is_using": external_knowledge_api_is_using, "count": count}, 200
return UsageCountResponse(is_using=external_knowledge_api_is_using, count=count).model_dump(mode="json"), 200
@console_ns.route("/datasets/external")
@@ -330,7 +315,9 @@ 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", dataset_detail_model)
@console_ns.response(
201, "External dataset created successfully", console_ns.models[DatasetDetailResponse.__name__]
)
@console_ns.response(400, "Invalid parameters")
@console_ns.response(403, "Permission denied")
@setup_required
@@ -360,17 +347,16 @@ class ExternalDatasetCreateApi(Resource):
except services.errors.dataset.DatasetNameDuplicateError:
raise DatasetNameDuplicateError()
item = marshal(dataset, dataset_detail_fields)
dataset_id_str = item["id"]
dataset_id_str = str(dataset.id)
permission_keys_map = enterprise_rbac_service.RBACService.DatasetPermissions.batch_get(
str(current_tenant_id),
current_user.id,
[dataset_id_str],
session=session,
)
item["permission_keys"] = permission_keys_map.get(dataset_id_str, [])
return item, 201
data = DatasetDetailResponse.model_validate(dataset).model_dump(mode="json")
data["permission_keys"] = permission_keys_map.get(dataset_id_str, [])
return data, 201
@console_ns.route("/datasets/<uuid:dataset_id>/external-hit-testing")
@@ -382,7 +368,7 @@ class ExternalKnowledgeHitTestingApi(Resource):
@console_ns.response(
200,
"External hit testing completed successfully",
console_ns.models[ExternalRetrievalTestResponse.__name__],
console_ns.models[ExternalHitTestingResponse.__name__],
)
@console_ns.response(404, "Dataset not found")
@console_ns.response(400, "Invalid parameters")
@@ -394,12 +380,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, db.session())
dataset = DatasetService.get_dataset(dataset_id_str, session)
if dataset is None:
raise NotFound("Dataset not found.")
try:
DatasetService.check_dataset_permission(dataset, current_user, db.session())
DatasetService.check_dataset_permission(dataset, current_user, session)
except services.errors.account.NoPermissionError as e:
raise Forbidden(str(e))
@@ -416,7 +402,7 @@ class ExternalKnowledgeHitTestingApi(Resource):
metadata_filtering_conditions=payload.metadata_filtering_conditions,
)
return response
return dump_response(ExternalHitTestingResponse, response)
except Exception as e:
raise InternalServerError(str(e))
@@ -427,11 +413,7 @@ 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[ExternalRetrievalTestResponse.__name__],
)
@console_ns.response(200, "Bedrock retrieval test completed", console_ns.models[BedrockRetrievalResponse.__name__])
def post(self):
payload = BedrockRetrievalPayload.model_validate(console_ns.payload or {})
@@ -439,4 +421,4 @@ class BedrockRetrievalApi(Resource):
result = ExternalDatasetTestService.knowledge_retrieval(
payload.retrieval_setting, payload.query, payload.knowledge_id
)
return result, 200
return dump_response(BedrockRetrievalResponse, 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(
db.session(), dataset_id_str, metadata_args, current_user, current_tenant_id
dataset_id_str, metadata_args, current_user, current_tenant_id, session=db.session()
)
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(db.session(), dataset)
metadata = MetadataService.get_dataset_metadatas(dataset, session=db.session())
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(
db.session(), dataset_id_str, metadata_id_str, name, current_user, current_tenant_id
dataset_id_str, metadata_id_str, name, current_user, current_tenant_id, session=db.session()
)
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(db.session(), dataset_id_str, metadata_id_str)
MetadataService.delete_metadata(dataset_id_str, metadata_id_str, session=db.session())
# 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(db.session(), dataset)
MetadataService.enable_built_in_field(dataset, session=db.session())
case "disable":
MetadataService.disable_built_in_field(db.session(), dataset)
MetadataService.disable_built_in_field(dataset, session=db.session())
# 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(db.session(), dataset, metadata_args, current_user)
MetadataService.update_documents_metadata(dataset, metadata_args, current_user, session=db.session())
# 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
@get_rag_pipeline
@with_session
@get_rag_pipeline
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
@get_rag_pipeline
@with_session
@get_rag_pipeline
def post(self, session: Session, current_user: Account, pipeline: Pipeline):
"""
Run published workflow
+5 -3
View File
@@ -2,6 +2,7 @@ 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
@@ -22,9 +23,10 @@ def get_rag_pipeline[**P, R](view_func: Callable[P, R]) -> Callable[P, R]:
del kwargs["pipeline_id"]
pipeline = db.session.scalar(
select(Pipeline).where(Pipeline.id == pipeline_id, Pipeline.tenant_id == current_tenant_id).limit(1)
)
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)
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(db.session(), language_prefix),
RecommendedAppService.get_recommended_apps_and_categories(language_prefix, session=db.session()),
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(db.session(), language_prefix),
RecommendedAppService.get_learn_dify_apps(language_prefix, session=db.session()),
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(db.session(), str(app_id))
return RecommendedAppService.get_recommend_app_detail(str(app_id), session=db.session())
@@ -38,11 +38,7 @@ class SavedMessageListApi(InstalledAppResource):
args = SavedMessageListQuery.model_validate(request.args.to_dict())
pagination = SavedMessageService.pagination_by_last_id(
db.session(),
app_model,
current_user,
str(args.last_id) if args.last_id else None,
args.limit,
app_model, current_user, str(args.last_id) if args.last_id else None, args.limit, session=db.session()
)
adapter = TypeAdapter(SavedMessageItem)
items = [adapter.validate_python(message, from_attributes=True) for message in pagination.data]
@@ -65,7 +61,7 @@ class SavedMessageListApi(InstalledAppResource):
payload = SavedMessageCreatePayload.model_validate(console_ns.payload or {})
try:
SavedMessageService.save(db.session(), app_model, current_user, str(payload.message_id))
SavedMessageService.save(app_model, current_user, str(payload.message_id), session=db.session())
except MessageNotExistsError:
raise NotFound("Message Not Exists.")
@@ -88,6 +84,6 @@ class SavedMessageApi(InstalledAppResource):
if app_model.mode != "completion":
raise NotCompletionAppError()
SavedMessageService.delete(db.session(), app_model, current_user, message_id_str)
SavedMessageService.delete(app_model, current_user, message_id_str, session=db.session())
return "", 204
+5 -5
View File
@@ -431,7 +431,7 @@ class TrialAppWorkflowRunApi(TrialAppResource):
invoke_from=InvokeFrom.EXPLORE,
streaming=True,
)
RecommendedAppService.add_trial_app_record(session, app_id, user_id)
RecommendedAppService.add_trial_app_record(app_id, user_id, session=session)
# 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(session, app_id, user_id)
RecommendedAppService.add_trial_app_record(app_id, user_id, session=session)
# 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(db.session(), app_id, user_id)
RecommendedAppService.add_trial_app_record(app_id, user_id, session=db.session())
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(db.session(), app_id, user_id)
RecommendedAppService.add_trial_app_record(app_id, user_id, session=db.session())
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(session, app_id, user_id)
RecommendedAppService.add_trial_app_record(app_id, user_id, session=session)
# response-contract:ignore compact_generate_response
return helper.compact_generate_response(response)
except services.errors.conversation.ConversationNotExistsError:
+9 -7
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(db.session(), current_tenant_id),
APIBasedExtensionService.get_all_by_tenant_id(current_tenant_id, session=db.session()),
)
@console_ns.doc("create_api_based_extension")
@@ -133,7 +133,7 @@ class APIBasedExtensionAPI(Resource):
api_key=payload.api_key,
)
extension = APIBasedExtensionService.save(db.session(), extension_data)
extension = APIBasedExtensionService.save(extension_data, session=db.session())
return APIBasedExtensionResponse(
id=extension.id,
name=extension.name,
@@ -158,7 +158,9 @@ class APIBasedExtensionDetailAPI(Resource):
return dump_response(
APIBasedExtensionResponse,
APIBasedExtensionService.get_with_tenant_id(db.session(), current_tenant_id, api_based_extension_id),
APIBasedExtensionService.get_with_tenant_id(
current_tenant_id, api_based_extension_id, session=db.session()
),
)
@console_ns.doc("update_api_based_extension")
@@ -174,7 +176,7 @@ class APIBasedExtensionDetailAPI(Resource):
api_based_extension_id = str(id)
extension_data_from_db = APIBasedExtensionService.get_with_tenant_id(
db.session(), current_tenant_id, api_based_extension_id
current_tenant_id, api_based_extension_id, session=db.session()
)
payload = APIBasedExtensionPayload.model_validate(console_ns.payload or {})
@@ -187,7 +189,7 @@ class APIBasedExtensionDetailAPI(Resource):
extension_data_from_db.api_key = payload.api_key
api_key_for_response = payload.api_key
APIBasedExtensionService.save(db.session(), extension_data_from_db)
APIBasedExtensionService.save(extension_data_from_db, session=db.session())
return APIBasedExtensionResponse(
id=extension_data_from_db.id,
name=extension_data_from_db.name,
@@ -208,9 +210,9 @@ class APIBasedExtensionDetailAPI(Resource):
api_based_extension_id = str(id)
extension_data_from_db = APIBasedExtensionService.get_with_tenant_id(
db.session(), current_tenant_id, api_based_extension_id
current_tenant_id, api_based_extension_id, session=db.session()
)
APIBasedExtensionService.delete(db.session(), extension_data_from_db)
APIBasedExtensionService.delete(extension_data_from_db, session=db.session())
return "", 204
+4 -4
View File
@@ -27,6 +27,7 @@ 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
@@ -117,8 +118,7 @@ class FileApi(Resource):
except services.errors.file.BlockedFileExtensionError as blocked_extension_error:
raise BlockedFileExtensionError(blocked_extension_error.description)
response = FileResponse.model_validate(upload_file, from_attributes=True)
return response.model_dump(mode="json"), 201
return dump_response(FileResponse, upload_file), 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 {"content": text}
return TextContentResponse(content=text).model_dump(mode="json")
@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 {"allowed_extensions": list(DOCUMENT_EXTENSIONS)}
return AllowedExtensionsResponse(allowed_extensions=list(DOCUMENT_EXTENSIONS)).model_dump(mode="json")
@@ -148,6 +148,8 @@ 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):
+2 -1
View File
@@ -69,7 +69,8 @@ def handle_user_connect(sid, data):
if not workflow_id:
return {"msg": "workflow_id is required"}, 400
result = collaboration_service.authorize_and_join_workflow_room(workflow_id, sid, session=db.session())
with sio.app.app_context():
result = collaboration_service.authorize_and_join_workflow_room(workflow_id, sid, session=db.session())
if not result:
return {"msg": "unauthorized"}, 401
+12 -4
View File
@@ -1,8 +1,9 @@
import logging
from collections.abc import Mapping
from typing import Any
from flask_restx import Resource
from pydantic import RootModel
from pydantic import Field, RootModel
from controllers.common.schema import register_response_schema_models
from controllers.console.wraps import (
@@ -10,6 +11,7 @@ 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
@@ -17,11 +19,17 @@ from . import console_ns
logger = logging.getLogger(__name__)
class SchemaDefinitionsResponse(RootModel[Any]):
root: Any
class SchemaDefinitionItemResponse(ResponseModel):
name: str
label: str
schema_: Mapping[str, Any] = Field(alias="schema")
register_response_schema_models(console_ns, SchemaDefinitionsResponse)
class SchemaDefinitionsResponse(RootModel[list[SchemaDefinitionItemResponse]]):
pass
register_response_schema_models(console_ns, SchemaDefinitionItemResponse, 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(db.session(), param.type, current_tenant_id, param.keyword)
tags = TagService.get_tags(param.type, current_tenant_id, param.keyword, session=db.session())
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(db.session(), email)
account = AccountService.get_account_by_email_with_case_fallback(email, session=db.session())
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 BinaryFileResponse, SimpleResultResponse
from controllers.common.fields import SimpleResultResponse, ValidationResultResponse
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,8 +22,7 @@ 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 graphon.model_runtime.utils.encoders import jsonable_encoder
from libs.helper import uuid_value
from libs.helper import dump_response, uuid_value
from libs.login import login_required
from models import Account
from services.billing_service import BillingService
@@ -92,13 +91,8 @@ class ModelProviderListResponse(ResponseModel):
data: list[ProviderResponse]
class ProviderCredentialResponse(ResponseModel):
credentials: dict[str, Any] | None = Field(default=None)
class ProviderCredentialValidateResponse(ResponseModel):
result: Literal["success", "error"]
error: str | None = None
class ProviderCredentialsResponse(ResponseModel):
credentials: dict[str, Any] | None = None
class ModelProviderPaymentCheckoutUrlResponse(ResponseModel):
@@ -118,19 +112,20 @@ 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, "Success", console_ns.models[ModelProviderListResponse.__name__])
@console_ns.response(
200, "Model providers retrieved successfully", console_ns.models[ModelProviderListResponse.__name__]
)
@setup_required
@login_required
@account_initialization_required
@@ -142,13 +137,17 @@ 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 jsonable_encoder({"data": provider_list})
return ModelProviderListResponse(data=provider_list).model_dump(mode="json")
@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, "Success", console_ns.models[ProviderCredentialResponse.__name__])
@console_ns.response(
200,
"Provider credentials retrieved successfully",
console_ns.models[ProviderCredentialsResponse.__name__],
)
@setup_required
@login_required
@account_initialization_required
@@ -163,7 +162,7 @@ class ModelProviderCredentialApi(Resource):
tenant_id=tenant_id, provider=provider, credential_id=args.credential_id
)
return {"credentials": credentials}
return ProviderCredentialsResponse(credentials=credentials).model_dump(mode="json")
@console_ns.expect(console_ns.models[ParserCredentialCreate.__name__])
@console_ns.response(201, "Credential created successfully", console_ns.models[SimpleResultResponse.__name__])
@@ -189,7 +188,7 @@ class ModelProviderCredentialApi(Resource):
except CredentialsValidateFailedError as ex:
raise ValueError(str(ex))
return {"result": "success"}, 201
return SimpleResultResponse(result="success").model_dump(mode="json"), 201
@console_ns.expect(console_ns.models[ParserCredentialUpdate.__name__])
@console_ns.response(200, "Credential updated successfully", console_ns.models[SimpleResultResponse.__name__])
@@ -216,7 +215,7 @@ class ModelProviderCredentialApi(Resource):
except CredentialsValidateFailedError as ex:
raise ValueError(str(ex))
return {"result": "success"}
return SimpleResultResponse(result="success").model_dump(mode="json")
@console_ns.expect(console_ns.models[ParserCredentialDelete.__name__])
@console_ns.response(204, "Credential deleted successfully")
@@ -258,7 +257,7 @@ class ModelProviderCredentialSwitchApi(Resource):
provider=provider,
credential_id=args.credential_id,
)
return {"result": "success"}
return SimpleResultResponse(result="success").model_dump(mode="json")
@console_ns.route("/workspaces/current/model-providers/<path:provider>/credentials/validate")
@@ -266,8 +265,8 @@ class ModelProviderValidateApi(Resource):
@console_ns.expect(console_ns.models[ParserCredentialValidate.__name__])
@console_ns.response(
200,
"Credential validation result",
console_ns.models[ProviderCredentialValidateResponse.__name__],
"Provider credentials validated successfully",
console_ns.models[ValidationResultResponse.__name__],
)
@setup_required
@login_required
@@ -292,12 +291,10 @@ class ModelProviderValidateApi(Resource):
result = False
error = str(ex)
response = {"result": "success" if result else "error"}
if not result:
response["error"] = error or "Unknown error"
return ValidationResultResponse(result="error", error=error or "Unknown error").model_dump(mode="json")
return response
return ValidationResultResponse(result="success").model_dump(mode="json")
@console_ns.route("/workspaces/<string:tenant_id>/model-providers/<path:provider>/<string:icon_type>/<string:lang>")
@@ -306,8 +303,9 @@ class ModelProviderIconApi(Resource):
Get model provider icon
"""
@console_ns.response(200, "Success", console_ns.models[BinaryFileResponse.__name__])
@console_ns.response(200, "Model provider icon")
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,
@@ -339,12 +337,16 @@ class PreferredProviderTypeUpdateApi(Resource):
tenant_id=tenant_id, provider=provider, preferred_provider_type=args.preferred_provider_type
)
return {"result": "success"}
return SimpleResultResponse(result="success").model_dump(mode="json")
@console_ns.route("/workspaces/current/model-providers/<path:provider>/checkout-url")
class ModelProviderPaymentCheckoutUrlApi(Resource):
@console_ns.response(200, "Success", console_ns.models[ModelProviderPaymentCheckoutUrlResponse.__name__])
@console_ns.response(
200,
"Model provider checkout URL retrieved successfully",
console_ns.models[ModelProviderPaymentCheckoutUrlResponse.__name__],
)
@setup_required
@login_required
@account_initialization_required
@@ -353,11 +355,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(db.session(), current_user)
BillingService.is_tenant_owner_or_admin(current_user, session=db.session())
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 data
return dump_response(ModelProviderPaymentCheckoutUrlResponse, data)
+74 -57
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
from controllers.common.fields import SimpleResultResponse, ValidationResultResponse
from controllers.common.schema import (
query_params_from_model,
register_enum_models,
@@ -28,7 +28,6 @@ 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
@@ -63,7 +62,7 @@ class ParserDeleteModels(BaseModel):
class LoadBalancingPayload(BaseModel):
configs: list[dict[str, Any]] | None = Field(default=None)
configs: list[dict[str, Any]] | None = None
enabled: bool | None = None
@@ -140,33 +139,38 @@ class DefaultModelDataResponse(ResponseModel):
data: DefaultModelResponse | None = None
class ModelWithProviderListResponse(ResponseModel):
class ProviderModelListResponse(ResponseModel):
data: list[ModelWithProviderEntityResponse]
class ProviderWithModelsDataResponse(ResponseModel):
class AvailableModelListResponse(ResponseModel):
data: list[ProviderWithModelsResponse]
class ModelCredentialLoadBalancingResponse(ResponseModel):
class ModelLoadBalancingConfigResponse(ResponseModel):
id: str
name: str
credentials: dict[str, Any]
credential_id: str | None = None
enabled: bool
configs: list[dict[str, Any]] = Field(default_factory=list)
in_cooldown: bool
ttl: int
class ModelLoadBalancingResponse(ResponseModel):
enabled: bool
configs: list[ModelLoadBalancingConfigResponse]
class ModelCredentialResponse(ResponseModel):
credentials: dict[str, Any] = Field(default_factory=dict)
credentials: dict[str, Any]
current_credential_id: str | None = None
current_credential_name: str | None = None
load_balancing: ModelCredentialLoadBalancingResponse
load_balancing: ModelLoadBalancingResponse
available_credentials: list[CredentialConfiguration]
class ModelCredentialValidateResponse(ResponseModel):
result: str
error: str | None = None
class ModelParameterRulesResponse(ResponseModel):
class ModelParameterRuleListResponse(ResponseModel):
data: list[ParameterRule]
@@ -187,12 +191,12 @@ register_schema_models(
register_response_schema_models(
console_ns,
SimpleResultResponse,
ValidationResultResponse,
DefaultModelDataResponse,
ModelWithProviderListResponse,
ProviderWithModelsDataResponse,
ProviderModelListResponse,
ModelCredentialResponse,
ModelCredentialValidateResponse,
ModelParameterRulesResponse,
ModelParameterRuleListResponse,
AvailableModelListResponse,
)
register_enum_models(console_ns, ModelType)
@@ -201,7 +205,9 @@ 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, "Success", console_ns.models[DefaultModelDataResponse.__name__])
@console_ns.response(
200, "Default model retrieved successfully", console_ns.models[DefaultModelDataResponse.__name__]
)
@setup_required
@login_required
@account_initialization_required
@@ -214,7 +220,7 @@ class DefaultModelApi(Resource):
tenant_id=tenant_id, model_type=args.model_type
)
return jsonable_encoder({"data": default_model_entity})
return DefaultModelDataResponse(data=default_model_entity).model_dump(mode="json")
@console_ns.expect(console_ns.models[ParserPostDefault.__name__])
@console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
@@ -247,12 +253,14 @@ class DefaultModelApi(Resource):
)
raise ex
return {"result": "success"}
return SimpleResultResponse(result="success").model_dump(mode="json")
@console_ns.route("/workspaces/current/model-providers/<path:provider>/models")
class ModelProviderModelApi(Resource):
@console_ns.response(200, "Success", console_ns.models[ModelWithProviderListResponse.__name__])
@console_ns.response(
200, "Provider models retrieved successfully", console_ns.models[ProviderModelListResponse.__name__]
)
@setup_required
@login_required
@account_initialization_required
@@ -261,10 +269,10 @@ class ModelProviderModelApi(Resource):
model_provider_service = ModelProviderService()
models = model_provider_service.get_models_by_provider(tenant_id=tenant_id, provider=provider)
return jsonable_encoder({"data": models})
return ProviderModelListResponse(data=models).model_dump(mode="json")
@console_ns.expect(console_ns.models[ParserPostModels.__name__])
@console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
@console_ns.response(200, "Model updated successfully", console_ns.models[SimpleResultResponse.__name__])
@setup_required
@login_required
@is_admin_or_owner_required
@@ -310,7 +318,7 @@ class ModelProviderModelApi(Resource):
tenant_id=tenant_id, provider=provider, model=args.model, model_type=args.model_type
)
return {"result": "success"}, 200
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
@console_ns.expect(console_ns.models[ParserDeleteModels.__name__])
@console_ns.response(204, "Model deleted successfully")
@@ -334,7 +342,11 @@ 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, "Success", console_ns.models[ModelCredentialResponse.__name__])
@console_ns.response(
200,
"Model credentials retrieved successfully",
console_ns.models[ModelCredentialResponse.__name__],
)
@setup_required
@login_required
@account_initialization_required
@@ -379,22 +391,23 @@ class ModelProviderModelCredentialApi(Resource):
model=args.model,
)
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,
}
)
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")
@console_ns.expect(console_ns.models[ParserCreateCredential.__name__])
@console_ns.response(201, "Credential created successfully", console_ns.models[SimpleResultResponse.__name__])
@console_ns.response(201, "Model credential created successfully", console_ns.models[SimpleResultResponse.__name__])
@setup_required
@login_required
@is_admin_or_owner_required
@@ -424,10 +437,10 @@ class ModelProviderModelCredentialApi(Resource):
)
raise ValueError(str(ex))
return {"result": "success"}, 201
return SimpleResultResponse(result="success").model_dump(mode="json"), 201
@console_ns.expect(console_ns.models[ParserUpdateCredential.__name__])
@console_ns.response(200, "Credential updated successfully", console_ns.models[SimpleResultResponse.__name__])
@console_ns.response(200, "Model credential updated successfully", console_ns.models[SimpleResultResponse.__name__])
@setup_required
@login_required
@is_admin_or_owner_required
@@ -452,7 +465,7 @@ class ModelProviderModelCredentialApi(Resource):
except CredentialsValidateFailedError as ex:
raise ValueError(str(ex))
return {"result": "success"}
return SimpleResultResponse(result="success").model_dump(mode="json")
@console_ns.expect(console_ns.models[ParserDeleteCredential.__name__])
@console_ns.response(204, "Credential deleted successfully")
@@ -498,7 +511,7 @@ class ModelProviderModelCredentialSwitchApi(Resource):
model=args.model,
credential_id=args.credential_id,
)
return {"result": "success"}
return SimpleResultResponse(result="success").model_dump(mode="json")
@console_ns.route(
@@ -520,7 +533,7 @@ class ModelProviderModelEnableApi(Resource):
tenant_id=tenant_id, provider=provider, model=args.model, model_type=args.model_type
)
return {"result": "success"}
return SimpleResultResponse(result="success").model_dump(mode="json")
@console_ns.route(
@@ -542,7 +555,7 @@ class ModelProviderModelDisableApi(Resource):
tenant_id=tenant_id, provider=provider, model=args.model, model_type=args.model_type
)
return {"result": "success"}
return SimpleResultResponse(result="success").model_dump(mode="json")
class ParserValidate(BaseModel):
@@ -559,8 +572,8 @@ class ModelProviderModelValidateApi(Resource):
@console_ns.expect(console_ns.models[ParserValidate.__name__])
@console_ns.response(
200,
"Credential validation result",
console_ns.models[ModelCredentialValidateResponse.__name__],
"Model credentials validated successfully",
console_ns.models[ValidationResultResponse.__name__],
)
@setup_required
@login_required
@@ -586,18 +599,20 @@ class ModelProviderModelValidateApi(Resource):
result = False
error = str(ex)
response = {"result": "success" if result else "error"}
if not result:
response["error"] = error or ""
return ValidationResultResponse(result="error", error=error or "").model_dump(mode="json")
return response
return ValidationResultResponse(result="success").model_dump(mode="json")
@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, "Success", console_ns.models[ModelParameterRulesResponse.__name__])
@console_ns.response(
200,
"Model parameter rules retrieved successfully",
console_ns.models[ModelParameterRuleListResponse.__name__],
)
@setup_required
@login_required
@account_initialization_required
@@ -610,12 +625,14 @@ class ModelProviderModelParameterRuleApi(Resource):
tenant_id=tenant_id, provider=provider, model=args.model
)
return jsonable_encoder({"data": parameter_rules})
return ModelParameterRuleListResponse(data=parameter_rules).model_dump(mode="json")
@console_ns.route("/workspaces/current/models/model-types/<string:model_type>")
class ModelProviderAvailableModelApi(Resource):
@console_ns.response(200, "Success", console_ns.models[ProviderWithModelsDataResponse.__name__])
@console_ns.response(
200, "Available models retrieved successfully", console_ns.models[AvailableModelListResponse.__name__]
)
@setup_required
@login_required
@account_initialization_required
@@ -624,4 +641,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 jsonable_encoder({"data": models})
return AvailableModelListResponse(data=models).model_dump(mode="json")
+27 -17
View File
@@ -31,7 +31,15 @@ from controllers.console.wraps import (
with_current_user_id,
)
from core.helper.position_helper import is_filtered
from core.plugin.entities.plugin import PluginCategory, PluginInstallationSource
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.impl.exc import PluginDaemonClientSideError
from core.plugin.plugin_service import PluginService
from core.tools.builtin_tool.providers._positions import BuiltinToolProviderSort
@@ -299,12 +307,12 @@ class PluginCategoryListResponse(ResponseModel):
has_more: bool
class PluginDaemonOperationResponse(RootModel[Any]):
root: Any
class PluginBundleUploadResponse(RootModel[list[PluginBundleDependency]]):
pass
class PluginListResponse(ResponseModel):
plugins: Any
plugins: list[PluginEntity]
total: int
@@ -334,15 +342,15 @@ class PluginInstallationsResponse(ResponseModel):
class PluginManifestResponse(ResponseModel):
manifest: Any
manifest: PluginDeclaration
class PluginTasksResponse(ResponseModel):
tasks: Any
tasks: list[PluginInstallTask]
class PluginTaskResponse(ResponseModel):
task: Any
task: PluginInstallTask
class PluginPermissionResponse(ResponseModel):
@@ -351,7 +359,7 @@ class PluginPermissionResponse(ResponseModel):
class PluginDynamicOptionsResponse(ResponseModel):
options: Any
options: list[PluginParameterOption]
class PluginOperationSuccessResponse(ResponseModel):
@@ -398,10 +406,12 @@ register_response_schema_models(
PluginCategoryBuiltinToolResponse,
PluginCategoryInstalledPluginResponse,
PluginCategoryListResponse,
PluginDaemonOperationResponse,
PluginBundleUploadResponse,
PluginDecodeResponse,
PluginDebuggingKeyResponse,
PluginDynamicOptionsResponse,
PluginInstallationsResponse,
PluginInstallTaskStartResponse,
PluginListResponse,
PluginManifestResponse,
PluginOperationSuccessResponse,
@@ -635,7 +645,7 @@ class PluginAssetApi(Resource):
@console_ns.route("/workspaces/current/plugin/upload/pkg")
class PluginUploadFromPkgApi(Resource):
@console_ns.response(200, "Success", console_ns.models[PluginDaemonOperationResponse.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginDecodeResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -656,7 +666,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[PluginDaemonOperationResponse.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginDecodeResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -676,7 +686,7 @@ class PluginUploadFromGithubApi(Resource):
@console_ns.route("/workspaces/current/plugin/upload/bundle")
class PluginUploadFromBundleApi(Resource):
@console_ns.response(200, "Success", console_ns.models[PluginDaemonOperationResponse.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginBundleUploadResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -697,7 +707,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[PluginDaemonOperationResponse.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginInstallTaskStartResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -718,7 +728,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[PluginDaemonOperationResponse.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginInstallTaskStartResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -745,7 +755,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[PluginDaemonOperationResponse.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginInstallTaskStartResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -891,7 +901,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[PluginDaemonOperationResponse.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginInstallTaskStartResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@@ -914,7 +924,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[PluginDaemonOperationResponse.__name__])
@console_ns.response(200, "Success", console_ns.models[PluginInstallTaskStartResponse.__name__])
@setup_required
@login_required
@account_initialization_required
+5 -78
View File
@@ -1,12 +1,9 @@
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
@@ -37,7 +34,8 @@ from controllers.console.wraps import (
from core.plugin.entities.plugin import PluginDependency
from extensions.ext_database import db
from fields.base import ResponseModel
from libs.helper import dump_response, to_timestamp
from fields.snippet_fields import SnippetListItemResponse, SnippetPaginationResponse, SnippetResponse
from libs.helper import dump_response
from libs.login import login_required
from models import Account
from models.snippet import SnippetType
@@ -65,77 +63,6 @@ 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))
@@ -259,7 +186,7 @@ class CustomizedSnippetDetailApi(Resource):
@login_required
@account_initialization_required
@with_current_tenant_id
def get(self, current_tenant_id: str, snippet_id: str):
def get(self, current_tenant_id: str, snippet_id: UUID):
"""Get customized snippet details."""
snippet_service = _snippet_service()
snippet = snippet_service.get_snippet_by_id(
@@ -535,4 +462,4 @@ class CustomizedSnippetUseCountIncrementApi(Resource):
session.commit()
session.refresh(snippet)
return SnippetUseCountResponse(result="success", use_count=snippet.use_count).model_dump(mode="json"), 200
return {"result": "success", "use_count": snippet.use_count}, 200
File diff suppressed because it is too large Load Diff
@@ -1,20 +1,19 @@
import logging
from typing import Any, Literal
from typing import Any
from flask import make_response, redirect, request
from flask_restx import Resource
from pydantic import BaseModel, Field, RootModel, model_validator
from pydantic import BaseModel, 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 BinaryFileResponse, RedirectResponse, SimpleResultResponse
from controllers.common.fields import SimpleResultResponse
from controllers.common.schema import register_response_schema_models, register_schema_models
from core.entities.parameter_entities import AppSelectorScope, ModelSelectorScope, ToolSelectorScope
from core.entities.provider_entities import ProviderConfig
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,
@@ -24,7 +23,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 graphon.model_runtime.utils.encoders import jsonable_encoder
from libs.helper import dump_response
from libs.login import login_required
from models.account import Account
from models.provider_ids import TriggerProviderID
@@ -59,9 +58,9 @@ class TriggerSubscriptionBuilderVerifyPayload(BaseModel):
class TriggerSubscriptionBuilderUpdatePayload(BaseModel):
name: str | 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)
parameters: dict[str, Any] | None = None
properties: dict[str, Any] | None = None
credentials: dict[str, Any] | None = None
@model_validator(mode="after")
def check_at_least_one_field(self):
@@ -71,70 +70,23 @@ class TriggerSubscriptionBuilderUpdatePayload(BaseModel):
class TriggerOAuthClientPayload(BaseModel):
client_params: dict[str, Any] | None = Field(default=None)
client_params: dict[str, Any] | None = 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]]):
root: list[TriggerProviderApiEntity]
pass
class TriggerSubscriptionListResponse(RootModel[list[TriggerProviderSubscriptionApiEntity]]):
root: list[TriggerProviderSubscriptionApiEntity]
class TriggerProviderSubscriptionListResponse(RootModel[list[TriggerProviderSubscriptionApiEntity]]):
pass
class TriggerSubscriptionBuilderCreateResponse(ResponseModel):
subscription_builder: SubscriptionBuilderApiEntity
class TriggerSubscriptionBuilderVerifyResponse(ResponseModel):
class TriggerVerificationResponse(ResponseModel):
verified: bool
@@ -142,6 +94,26 @@ 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,
@@ -151,27 +123,24 @@ register_schema_models(
)
register_response_schema_models(
console_ns,
BinaryFileResponse,
RedirectResponse,
SimpleResultResponse,
TriggerOAuthAuthorizeResponse,
TriggerOAuthClientResponse,
TriggerProviderOpaqueResponse,
TriggerProviderApiEntity,
TriggerProviderErrorResponse,
TriggerProviderListResponse,
TriggerProviderSubscriptionApiEntity,
TriggerSubscriptionListResponse,
SubscriptionBuilderApiEntity,
TriggerProviderSubscriptionListResponse,
TriggerSubscriptionBuilderCreateResponse,
TriggerSubscriptionBuilderVerifyResponse,
RequestLog,
TriggerSubscriptionBuilderLogsResponse,
SubscriptionBuilderApiEntity,
TriggerVerificationResponse,
)
@console_ns.route("/workspaces/current/trigger-provider/<path:provider>/icon")
class TriggerProviderIconApi(Resource):
@console_ns.response(200, "Success", console_ns.models[BinaryFileResponse.__name__])
# response-contract:ignore binary trigger provider icon
@console_ns.response(200, "Trigger provider icon")
@setup_required
@login_required
@account_initialization_required
@@ -182,31 +151,45 @@ class TriggerProviderIconApi(Resource):
@console_ns.route("/workspaces/current/triggers")
class TriggerProviderListApi(Resource):
@console_ns.response(200, "Success", console_ns.models[TriggerProviderListResponse.__name__])
@console_ns.response(
200,
"Trigger providers retrieved successfully",
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 jsonable_encoder(TriggerProviderService.list_trigger_providers(tenant_id))
return dump_response(TriggerProviderListResponse, TriggerProviderService.list_trigger_providers(tenant_id))
@console_ns.route("/workspaces/current/trigger-provider/<path:provider>/info")
class TriggerProviderInfoApi(Resource):
@console_ns.response(200, "Success", console_ns.models[TriggerProviderApiEntity.__name__])
@console_ns.response(
200,
"Trigger provider retrieved successfully",
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"""
return jsonable_encoder(TriggerProviderService.get_trigger_provider(tenant_id, TriggerProviderID(provider)))
provider_entity = TriggerProviderService.get_trigger_provider(tenant_id, TriggerProviderID(provider))
return provider_entity.model_dump(mode="json")
@console_ns.route("/workspaces/current/trigger-provider/<path:provider>/subscriptions/list")
class TriggerSubscriptionListApi(Resource):
@console_ns.response(200, "Success", console_ns.models[TriggerSubscriptionListResponse.__name__])
@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__])
@setup_required
@login_required
@edit_permission_required
@@ -216,16 +199,18 @@ 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 jsonable_encoder(
return dump_response(
TriggerProviderSubscriptionListResponse,
TriggerProviderService.list_trigger_provider_subscriptions(
tenant_id=tenant_id,
provider_id=TriggerProviderID(provider),
user=user,
)
),
)
except ValueError as e:
return jsonable_encoder({"error": str(e)}), 404
return TriggerProviderErrorResponse(error=str(e)).model_dump(mode="json"), 404
except Exception as e:
logger.exception("Error listing trigger providers", exc_info=e)
raise
@@ -236,7 +221,11 @@ class TriggerSubscriptionListApi(Resource):
)
class TriggerSubscriptionBuilderCreateApi(Resource):
@console_ns.expect(console_ns.models[TriggerSubscriptionBuilderCreatePayload.__name__])
@console_ns.response(200, "Success", console_ns.models[TriggerSubscriptionBuilderCreateResponse.__name__])
@console_ns.response(
200,
"Trigger subscription builder created successfully",
console_ns.models[TriggerSubscriptionBuilderCreateResponse.__name__],
)
@setup_required
@login_required
@edit_permission_required
@@ -246,6 +235,7 @@ 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:
@@ -256,7 +246,9 @@ class TriggerSubscriptionBuilderCreateApi(Resource):
provider_id=TriggerProviderID(provider),
credential_type=credential_type,
)
return jsonable_encoder({"subscription_builder": subscription_builder})
return TriggerSubscriptionBuilderCreateResponse(subscription_builder=subscription_builder).model_dump(
mode="json"
)
except Exception as e:
logger.exception("Error adding provider credential", exc_info=e)
raise
@@ -266,7 +258,11 @@ class TriggerSubscriptionBuilderCreateApi(Resource):
"/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/<path:subscription_builder_id>",
)
class TriggerSubscriptionBuilderGetApi(Resource):
@console_ns.response(200, "Success", console_ns.models[SubscriptionBuilderApiEntity.__name__])
@console_ns.response(
200,
"Trigger subscription builder retrieved successfully",
console_ns.models[SubscriptionBuilderApiEntity.__name__],
)
@setup_required
@login_required
@edit_permission_required
@@ -274,9 +270,8 @@ class TriggerSubscriptionBuilderGetApi(Resource):
@account_initialization_required
def get(self, provider: str, subscription_builder_id: str):
"""Get a subscription instance for a trigger provider"""
return jsonable_encoder(
TriggerSubscriptionBuilderService.get_subscription_builder_by_id(subscription_builder_id)
)
subscription_builder = TriggerSubscriptionBuilderService.get_subscription_builder_by_id(subscription_builder_id)
return subscription_builder.model_dump(mode="json")
@console_ns.route(
@@ -284,7 +279,11 @@ class TriggerSubscriptionBuilderGetApi(Resource):
)
class TriggerSubscriptionBuilderVerifyApi(Resource):
@console_ns.expect(console_ns.models[TriggerSubscriptionBuilderVerifyPayload.__name__])
@console_ns.response(200, "Success", console_ns.models[TriggerSubscriptionBuilderVerifyResponse.__name__])
@console_ns.response(
200,
"Trigger subscription builder verified successfully",
console_ns.models[TriggerVerificationResponse.__name__],
)
@setup_required
@login_required
@edit_permission_required
@@ -294,11 +293,12 @@ 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
return TriggerSubscriptionBuilderService.update_and_verify_builder(
result = TriggerSubscriptionBuilderService.update_and_verify_builder(
tenant_id=tenant_id,
user_id=user.id,
provider_id=TriggerProviderID(provider),
@@ -307,6 +307,7 @@ 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
@@ -317,7 +318,11 @@ class TriggerSubscriptionBuilderVerifyApi(Resource):
)
class TriggerSubscriptionBuilderUpdateApi(Resource):
@console_ns.expect(console_ns.models[TriggerSubscriptionBuilderUpdatePayload.__name__])
@console_ns.response(200, "Success", console_ns.models[SubscriptionBuilderApiEntity.__name__])
@console_ns.response(
200,
"Trigger subscription builder updated successfully",
console_ns.models[SubscriptionBuilderApiEntity.__name__],
)
@setup_required
@login_required
@edit_permission_required
@@ -326,21 +331,20 @@ 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 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,
),
)
)
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")
except Exception as e:
logger.exception("Error updating provider credential", exc_info=e)
raise
@@ -350,7 +354,11 @@ class TriggerSubscriptionBuilderUpdateApi(Resource):
"/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/logs/<path:subscription_builder_id>",
)
class TriggerSubscriptionBuilderLogsApi(Resource):
@console_ns.response(200, "Success", console_ns.models[TriggerSubscriptionBuilderLogsResponse.__name__])
@console_ns.response(
200,
"Trigger subscription builder logs retrieved successfully",
console_ns.models[TriggerSubscriptionBuilderLogsResponse.__name__],
)
@setup_required
@login_required
@edit_permission_required
@@ -358,9 +366,10 @@ 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 jsonable_encoder({"logs": [log.model_dump(mode="json") for log in logs]})
return dump_response(TriggerSubscriptionBuilderLogsResponse, {"logs": logs})
except Exception as e:
logger.exception("Error getting request logs for subscription builder", exc_info=e)
raise
@@ -371,7 +380,9 @@ class TriggerSubscriptionBuilderLogsApi(Resource):
)
class TriggerSubscriptionBuilderBuildApi(Resource):
@console_ns.expect(console_ns.models[TriggerSubscriptionBuilderUpdatePayload.__name__])
@console_ns.response(200, "Success", console_ns.models[TriggerProviderOpaqueResponse.__name__])
@console_ns.response(
200, "Trigger subscription builder built successfully", console_ns.models[SimpleResultResponse.__name__]
)
@setup_required
@login_required
@edit_permission_required
@@ -395,7 +406,7 @@ class TriggerSubscriptionBuilderBuildApi(Resource):
properties=payload.properties,
),
)
return 200
return SimpleResultResponse(result="success").model_dump(mode="json")
except Exception as e:
logger.exception("Error building provider credential", exc_info=e)
raise ValueError(str(e)) from e
@@ -406,7 +417,9 @@ class TriggerSubscriptionBuilderBuildApi(Resource):
)
class TriggerSubscriptionUpdateApi(Resource):
@console_ns.expect(console_ns.models[TriggerSubscriptionBuilderUpdatePayload.__name__])
@console_ns.response(200, "Success", console_ns.models[TriggerProviderOpaqueResponse.__name__])
@console_ns.response(
200, "Trigger subscription updated successfully", console_ns.models[SimpleResultResponse.__name__]
)
@setup_required
@login_required
@edit_permission_required
@@ -415,6 +428,7 @@ 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(
@@ -440,7 +454,7 @@ class TriggerSubscriptionUpdateApi(Resource):
name=request.name,
properties=request.properties,
)
return 200
return SimpleResultResponse(result="success").model_dump(mode="json")
# For the rest cases(API_KEY, OAUTH2)
# we need to call third party provider(e.g. GitHub) to rebuild the subscription
@@ -452,7 +466,7 @@ class TriggerSubscriptionUpdateApi(Resource):
credentials=request.credentials or subscription.credentials,
parameters=request.parameters or subscription.parameters,
)
return 200
return SimpleResultResponse(result="success").model_dump(mode="json")
except ValueError as e:
raise BadRequest(str(e))
except Exception as e:
@@ -473,6 +487,7 @@ 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
@@ -487,7 +502,7 @@ class TriggerSubscriptionDeleteApi(Resource):
tenant_id=tenant_id,
subscription_id=subscription_id,
)
return {"result": "success"}
return SimpleResultResponse(result="success").model_dump(mode="json")
except ValueError as e:
raise BadRequest(str(e))
except Exception as e:
@@ -497,9 +512,10 @@ 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,
"Authorization URL retrieved successfully",
"Trigger OAuth authorization URL generated successfully",
console_ns.models[TriggerOAuthAuthorizeResponse.__name__],
)
@setup_required
@@ -509,10 +525,12 @@ 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(
@@ -556,15 +574,12 @@ class TriggerOAuthAuthorizeApi(Resource):
system_credentials=oauth_client_params,
)
# Create response with cookie
response = make_response(
jsonable_encoder(
{
"authorization_url": authorization_url_response.authorization_url,
"subscription_builder_id": subscription_builder.id,
"subscription_builder": subscription_builder,
}
)
TriggerOAuthAuthorizeResponse(
authorization_url=authorization_url_response.authorization_url,
subscription_builder_id=subscription_builder.id,
subscription_builder=subscription_builder,
).model_dump(mode="json")
)
response.set_cookie(
"context_id",
@@ -583,11 +598,8 @@ class TriggerOAuthAuthorizeApi(Resource):
@console_ns.route("/oauth/plugin/<path:provider>/trigger/callback")
class TriggerOAuthCallbackApi(Resource):
@console_ns.response(
302,
"Redirect to console OAuth callback page",
console_ns.models[RedirectResponse.__name__],
)
# response-contract:ignore redirect response
@console_ns.response(302, "Redirect to OAuth callback page")
@setup_required
def get(self, provider: str):
"""Handle OAuth callback for trigger provider"""
@@ -653,7 +665,11 @@ class TriggerOAuthCallbackApi(Resource):
@console_ns.route("/workspaces/current/trigger-provider/<path:provider>/oauth/client")
class TriggerOAuthClientManageApi(Resource):
@console_ns.response(200, "Success", console_ns.models[TriggerOAuthClientResponse.__name__])
@console_ns.response(
200,
"Trigger OAuth client retrieved successfully",
console_ns.models[TriggerOAuthClientResponse.__name__],
)
@setup_required
@login_required
@is_admin_or_owner_required
@@ -662,6 +678,7 @@ 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)
@@ -682,24 +699,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 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 {},
}
)
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")
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, "Success", console_ns.models[SimpleResultResponse.__name__])
@console_ns.response(
200, "Trigger OAuth client saved successfully", console_ns.models[SimpleResultResponse.__name__]
)
@setup_required
@login_required
@is_admin_or_owner_required
@@ -708,16 +725,18 @@ 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)
return TriggerProviderService.save_custom_oauth_client_params(
result = 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))
@@ -725,22 +744,26 @@ 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)
return TriggerProviderService.delete_custom_oauth_client_params(
result = 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:
@@ -753,7 +776,11 @@ class TriggerOAuthClientManageApi(Resource):
)
class TriggerSubscriptionVerifyApi(Resource):
@console_ns.expect(console_ns.models[TriggerSubscriptionBuilderVerifyPayload.__name__])
@console_ns.response(200, "Success", console_ns.models[TriggerSubscriptionBuilderVerifyResponse.__name__])
@console_ns.response(
200,
"Trigger subscription verified successfully",
console_ns.models[TriggerVerificationResponse.__name__],
)
@setup_required
@login_required
@edit_permission_required
@@ -763,6 +790,7 @@ 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:
@@ -773,7 +801,7 @@ class TriggerSubscriptionVerifyApi(Resource):
subscription_id=subscription_id,
credentials=verify_request.credentials,
)
return result
return dump_response(TriggerVerificationResponse, 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(db.session(), current_user.id)
for tenant, membership in TenantService.get_workspaces_for_account(current_user.id, session=db.session())
if tenant.status == TenantStatus.NORMAL
]
tenants = [tenant for tenant, _ in tenant_rows]
+2 -4
View File
@@ -1,5 +1,3 @@
from mimetypes import guess_extension
from flask import request
from flask_restx import Resource
from flask_restx.api import HTTPStatus
@@ -8,7 +6,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
from core.tools.tool_file_manager import ToolFileManager, resolve_extension
from core.workflow.file_reference import build_file_reference
from fields.file_fields import FileResponse
@@ -110,7 +108,7 @@ class PluginUploadFileApi(Resource):
conversation_id=args.conversation_id,
)
extension = guess_extension(tool_file.mimetype) or ".bin"
extension = resolve_extension(filename=tool_file.name, mimetype=tool_file.mimetype)
preview_url = ToolFileManager.sign_file(tool_file_id=tool_file.id, extension=extension)
# Create a dictionary with all the necessary attributes
@@ -476,6 +476,7 @@ 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={
+12 -9
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, ValidationError
from pydantic import BaseModel, Field, RootModel, ValidationError
from sqlalchemy import select
from sqlalchemy.orm import Session, sessionmaker
from controllers.common.schema import register_schema_model
from controllers.common.schema import register_response_schema_models, 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,7 +33,12 @@ 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")
@@ -42,13 +47,10 @@ 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.doc(
responses={
200: "MCP response successfully processed",
400: "Invalid MCP request or parameters",
404: "Server or app not found",
}
)
@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")
def post(self, server_code: str):
"""Handle MCP requests for a specific server.
@@ -64,6 +66,7 @@ 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,11 +2,13 @@ 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,6 +45,7 @@ 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>/describe.
"""`?fields=` allow-list for GET /apps/<id>.
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>/export."""
"""Query parameters for GET /apps/<app_id>/dsl."""
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>/form/human_input/<token>. `extra='forbid'`
"""Empty 200 body for POST /apps/<id>/human-input-forms/<token>:submit. `extra='forbid'`
pins `additionalProperties: false` so the generated contract is an exact `{}` rather
than an under-annotated open object."""
+69
View File
@@ -0,0 +1,69 @@
"""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
+8 -6
View File
@@ -45,8 +45,10 @@ 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(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 []
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 []
)
default_ws_id = _pick_default_workspace(memberships)
return AccountResponse(
@@ -63,7 +65,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(db.session(), redis_client, str(auth_data.token_id))
revoke_oauth_token(redis_client, str(auth_data.token_id), session=db.session())
return RevokeResponse(status="revoked")
@@ -81,7 +83,7 @@ class AccountSessionsApi(Resource):
page = query.page
limit = query.limit
all_rows = list_active_sessions(db.session(), ctx, now)
all_rows = list_active_sessions(ctx, now, session=db.session())
total = len(all_rows)
sliced = all_rows[(page - 1) * limit : page * limit]
@@ -117,10 +119,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(db.session(), session_id, ctx):
if not token_belongs_to_subject(session_id, ctx, session=db.session()):
raise NotFound("session not found")
revoke_oauth_token(db.session(), redis_client, session_id)
revoke_oauth_token(redis_client, session_id, session=db.session())
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 confirm endpoint.
(major version mismatch). Callers must then POST to the imports :confirm method.
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>/export")
@openapi_ns.route("/apps/<string:app_id>/dsl")
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>/check-dependencies")
@openapi_ns.route("/apps/<string:app_id>/dependencies:check")
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(db.session(), str(parsed_uuid))
app = AppService.get_visible_app_by_id(str(parsed_uuid), session=db.session())
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(db.session(), name=app_id, tenant_id=workspace_id)
matches = AppService.find_visible_apps_by_name(name=app_id, tenant_id=workspace_id, session=db.session())
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>/describe")
@openapi_ns.route("/apps/<string:app_id>")
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(db.session(), str(parsed_uuid))
app: App | None = AppService.get_visible_app_by_id(str(parsed_uuid), session=db.session())
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(db.session(), workspace_id)
tenant_name = TenantService.get_tenant_name(workspace_id, session=db.session())
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(db.session(), workspace_id)
tenant_name = TenantService.get_tenant_name(workspace_id, session=db.session())
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(db.session(), page_result.app_ids)
str(a.id): a for a in AppService.find_visible_apps_by_ids(page_result.app_ids, session=db.session())
}
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(db.session(), tenant_ids)}
tenants_by_id = {str(t.id): t for t in TenantService.get_tenants_by_ids(tenant_ids, session=db.session())}
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>/describe")
@openapi_ns.route("/permitted-external-apps/<string:app_id>")
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(db.session(), app_id)
app = AppService.get_app_by_id(app_id, session=db.session())
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(db.session(), str(data.app.tenant_id))
tenant = TenantService.get_tenant_by_id(str(data.app.tenant_id), session=db.session())
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(db.session(), workspace_id)
tenant = TenantService.get_tenant_by_id(workspace_id, session=db.session())
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(db.session(), str(data.account_id))
account = AccountService.get_account_by_id(str(data.account_id), session=db.session())
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(db.session(), str(data.account_id), str(data.tenant.id))
role = TenantService.get_account_role_in_tenant(str(data.account_id), str(data.tenant.id), session=db.session())
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(db.session(), data.account_id, data.tenant.id):
if not TenantService.account_belongs_to_tenant(data.account_id, data.tenant.id, session=db.session()):
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(db.session(), data.external_identity.email)
account = AccountService.get_account_by_email(data.external_identity.email, session=db.session())
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 — upload a file for use in app inputs."""
"""POST /openapi/v1/apps/<app_id>/files — 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/upload")
@openapi_ns.route("/apps/<string:app_id>/files")
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")
+6 -3
View File
@@ -1,8 +1,8 @@
"""
OpenAPI bearer-authed human input form endpoints.
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
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
"""
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>/form/human_input/<string:form_token>")
@openapi_ns.route("/apps/<string:app_id>/human-input-forms/<string:form_token>")
class OpenApiWorkflowHumanInputFormApi(Resource):
@openapi_ns.response(200, "Form definition", openapi_ns.models[HumanInputFormDefinitionResponse.__name__])
@auth_router.guard(
@@ -79,6 +79,9 @@ 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,7 +247,6 @@ 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,
@@ -256,6 +255,7 @@ 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(db.session(), str(account.id))
rows = TenantService.get_workspaces_for_account(str(account.id), session=db.session())
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(db.session(), claims.email):
if AccountService.has_active_account_with_email(claims.email, session=db.session()):
_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(db.session(), claims.subject_email):
if AccountService.has_active_account_with_email(claims.subject_email, session=db.session()):
_emit_external_rejection_audit(state, claims, reason="email_belongs_to_dify_account")
raise Forbidden("email_belongs_to_dify_account")
@@ -293,7 +293,6 @@ 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,
@@ -302,6 +301,7 @@ 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
+13 -21
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(db.session(), workspace_id)
tenant = TenantService.get_tenant_by_id(workspace_id, session=db.session())
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(db.session(), str(account_id)) if account_id else None
account = AccountService.get_account_by_id(str(account_id), session=db.session()) 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(db.session(), str(auth_data.account_id))
rows = TenantService.get_workspaces_for_account(str(auth_data.account_id), session=db.session())
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(db.session(), str(auth_data.account_id), workspace_id)
row = TenantService.find_workspace_for_account(str(auth_data.account_id), workspace_id, session=db.session())
# 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(db.session(), str(auth_data.account_id), workspace_id)
row = TenantService.find_workspace_for_account(str(auth_data.account_id), workspace_id, session=db.session())
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(db.session(), normalized_email)
member = AccountService.get_account_by_email_with_case_fallback(normalized_email, session=db.session())
if member is None:
# invite_new_member just created or fetched this account.
raise RuntimeError("invited member missing from DB after invite")
@@ -212,11 +212,12 @@ class WorkspaceMembersApi(Resource):
@openapi_ns.route("/workspaces/<string:workspace_id>/members/<string:member_id>")
class WorkspaceMemberApi(Resource):
"""Remove a member.
"""Remove a member (DELETE) or change a member's role (PATCH).
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.
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.
"""
@auth_router.guard_workspace(
@@ -228,7 +229,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(db.session(), member_id)
member = AccountService.get_account_by_id(member_id, session=db.session())
if member is None:
raise NotFound("member not found")
@@ -243,15 +244,6 @@ 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}),
@@ -259,10 +251,10 @@ class WorkspaceMemberRoleApi(Resource):
)
@returns(200, MemberActionResponse, description="Role updated")
@accepts(body=MemberRoleUpdatePayload)
def put(self, workspace_id: str, member_id: str, *, auth_data: AuthData, body: MemberRoleUpdatePayload):
def patch(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(db.session(), member_id)
member = AccountService.get_account_by_id(member_id, session=db.session())
if member is None:
raise NotFound("member not found")
+26 -20
View File
@@ -12,8 +12,13 @@ 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, AnnotationList
from fields.base import ResponseModel
from fields.annotation_fields import (
Annotation,
AnnotationJobStatusDetailResponse,
AnnotationJobStatusResponse,
AnnotationList,
)
from libs.helper import dump_response
from models.model import App
from services.annotation_service import (
AppAnnotationService,
@@ -46,12 +51,6 @@ 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"],
@@ -67,7 +66,13 @@ register_schema_models(
Annotation,
AnnotationList,
)
register_response_schema_models(service_api_ns, AnnotationJobStatusResponse)
register_response_schema_models(
service_api_ns,
Annotation,
AnnotationList,
AnnotationJobStatusResponse,
AnnotationJobStatusDetailResponse,
)
@service_api_ns.route("/apps/annotation-reply/<string:action>")
@@ -113,7 +118,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 result, 200
return dump_response(AnnotationJobStatusResponse, result), 200
@service_api_ns.route("/apps/annotation-reply/<string:action>/status/<uuid:job_id>")
@@ -151,7 +156,7 @@ class AnnotationReplyActionStatusApi(Resource):
@service_api_ns.response(
200,
"Job status retrieved successfully",
service_api_ns.models[AnnotationJobStatusResponse.__name__],
service_api_ns.models[AnnotationJobStatusDetailResponse.__name__],
)
@validate_app_token
def get(self, app_model: App, job_id: UUID, action: str):
@@ -166,9 +171,13 @@ class AnnotationReplyActionStatusApi(Resource):
error_msg = ""
if job_status == "error":
app_annotation_error_key = f"{action}_app_annotation_error_{job_id_str}"
error_msg = redis_client.get(app_annotation_error_key).decode()
error_result = redis_client.get(app_annotation_error_key)
if error_result is not None:
error_msg = error_result.decode()
return {"job_id": job_id_str, "job_status": job_status, "error_msg": error_msg}, 200
return AnnotationJobStatusDetailResponse(
job_id=job_id_str, job_status=job_status, error_msg=error_msg
).model_dump(mode="json"), 200
@service_api_ns.route("/apps/annotations")
@@ -204,14 +213,13 @@ 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)
response = AnnotationList(
return AnnotationList(
data=annotation_models,
has_more=len(annotation_list) == query.limit,
limit=query.limit,
total=total,
page=query.page,
)
return response.model_dump(mode="json")
).model_dump(mode="json")
@service_api_ns.doc(
summary="Create Annotation",
@@ -246,8 +254,7 @@ class AnnotationListApi(Resource):
annotation = AppAnnotationService.insert_app_annotation_directly(
insert_args, app_model.id, session=db.session()
)
response = Annotation.model_validate(annotation, from_attributes=True)
return response.model_dump(mode="json"), HTTPStatus.CREATED
return dump_response(Annotation, annotation), HTTPStatus.CREATED
@service_api_ns.route("/apps/annotations/<uuid:annotation_id>")
@@ -288,8 +295,7 @@ 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())
response = Annotation.model_validate(annotation, from_attributes=True)
return response.model_dump(mode="json")
return dump_response(Annotation, annotation)
@service_api_ns.doc(
summary="Delete Annotation",
+4 -4
View File
@@ -25,6 +25,7 @@ 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
@@ -102,7 +103,7 @@ class AudioApi(Resource):
try:
response = AudioService.transcript_asr(app_model=app_model, file=file, end_user=end_user.id)
return response
return dump_response(AudioTranscriptResponse, response)
except services.errors.app_model_config.AppModelConfigBrokenError:
logger.exception("App model config broken.")
raise AppUnavailableError()
@@ -165,6 +166,7 @@ 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):
@@ -186,7 +188,7 @@ class TextApi(Resource):
message_id,
end_user_id=end_user.id,
)
response = AudioService.transcript_tts(
return AudioService.transcript_tts(
app_model=app_model,
session=db.session(),
text=text,
@@ -194,8 +196,6 @@ 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()
+8 -14
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 GeneratedAppResponse, SimpleResultResponse
from controllers.common.fields import 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, GeneratedAppResponse, SimpleResultResponse)
register_response_schema_models(service_api_ns, SimpleResultResponse)
@service_api_ns.route("/completion-messages")
@@ -201,11 +201,7 @@ class CompletionApi(Resource):
500: "Internal server error",
}
)
@service_api_ns.response(
200,
"Completion created successfully",
service_api_ns.models[GeneratedAppResponse.__name__],
)
@service_api_ns.response(200, "Completion created successfully")
@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):
@@ -242,6 +238,7 @@ 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.")
@@ -304,7 +301,7 @@ class CompletionStopApi(Resource):
app_mode=AppMode.value_of(app_model.mode),
)
return {"result": "success"}, 200
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
@service_api_ns.route("/chat-messages")
@@ -354,11 +351,7 @@ class ChatApi(Resource):
500: "Internal server error",
}
)
@service_api_ns.response(
200,
"Message sent successfully",
service_api_ns.models[GeneratedAppResponse.__name__],
)
@service_api_ns.response(200, "Message sent successfully")
@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):
@@ -393,6 +386,7 @@ 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))
@@ -464,4 +458,4 @@ class ChatStopApi(Resource):
app_mode=app_mode,
)
return {"result": "success"}, 200
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
+11 -22
View File
@@ -24,7 +24,7 @@ from fields.conversation_fields import (
SimpleConversation,
)
from graphon.variables.types import SegmentType
from libs.helper import UUIDStrOrEmpty, to_timestamp
from libs.helper import UUIDStrOrEmpty, dump_response, to_timestamp
from models.model import App, AppMode, EndUser
from services.conversation_service import ConversationService
@@ -142,15 +142,13 @@ register_schema_models(
ConversationRenamePayload,
ConversationVariablesQuery,
ConversationVariableUpdatePayload,
ConversationVariableResponse,
ConversationVariableInfiniteScrollPaginationResponse,
)
register_response_schema_models(
service_api_ns,
ConversationInfiniteScrollPagination,
SimpleConversation,
ConversationVariableResponse,
ConversationVariableInfiniteScrollPaginationResponse,
ConversationInfiniteScrollPagination,
SimpleConversation,
)
@@ -166,9 +164,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",
@@ -192,7 +190,7 @@ class ConversationApi(Resource):
raise NotChatAppError()
query_args = ConversationListQuery.model_validate(request.args.to_dict())
last_id = str(query_args.last_id) if query_args.last_id else None
last_id = query_args.last_id or None
try:
with sessionmaker(db.engine).begin() as session:
@@ -208,9 +206,7 @@ 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.")
@@ -301,11 +297,7 @@ class ConversationRenameApi(Resource):
conversation = ConversationService.rename(
app_model, conversation_id, end_user, payload.name, payload.auto_generate, session=db.session()
)
return (
TypeAdapter(SimpleConversation)
.validate_python(conversation, from_attributes=True)
.model_dump(mode="json")
)
return dump_response(SimpleConversation, conversation)
except services.errors.conversation.ConversationNotExistsError:
raise NotFound("Conversation Not Exists.")
@@ -322,10 +314,9 @@ 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."})
@service_api_ns.doc(params={"c_id": "Conversation ID.", **query_params_from_model(ConversationVariablesQuery)})
@service_api_ns.doc(
responses={
200: "Variables retrieved successfully",
@@ -352,7 +343,7 @@ class ConversationVariablesApi(Resource):
conversation_id = str(c_id)
query_args = ConversationVariablesQuery.model_validate(request.args.to_dict())
last_id = str(query_args.last_id) if query_args.last_id else None
last_id = query_args.last_id or None
try:
pagination = ConversationService.get_conversational_variable(
@@ -364,9 +355,7 @@ class ConversationVariablesApi(Resource):
query_args.variable_name,
session=db.session(),
)
return ConversationVariableInfiniteScrollPaginationResponse.model_validate(
pagination, from_attributes=True
).model_dump(mode="json")
return dump_response(ConversationVariableInfiniteScrollPaginationResponse, pagination)
except services.errors.conversation.ConversationNotExistsError:
raise NotFound("Conversation Not Exists.")
@@ -425,7 +414,7 @@ class ConversationVariableDetailApi(Resource):
variable = ConversationService.update_conversation_variable(
app_model, conversation_id, variable_id_str, end_user, payload.value, session=db.session()
)
return ConversationVariableResponse.model_validate(variable, from_attributes=True).model_dump(mode="json")
return dump_response(ConversationVariableResponse, variable)
except services.errors.conversation.ConversationNotExistsError:
raise NotFound("Conversation Not Exists.")
except services.errors.conversation.ConversationVariableNotExistsError:
+2 -2
View File
@@ -16,6 +16,7 @@ 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
@@ -87,5 +88,4 @@ class FileApi(Resource):
except services.errors.file.UnsupportedFileTypeError:
raise UnsupportedFileTypeError()
response = FileResponse.model_validate(upload_file, from_attributes=True)
return response.model_dump(mode="json"), 201
return dump_response(FileResponse, upload_file), 201
+17 -18
View File
@@ -60,12 +60,14 @@ 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=(
@@ -76,15 +78,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",
}
@@ -105,8 +107,8 @@ class MessageListApi(Resource):
raise NotChatAppError()
query_args = MessageListQuery.model_validate(request.args.to_dict())
conversation_id = str(query_args.conversation_id)
first_id = str(query_args.first_id) if query_args.first_id else None
conversation_id = query_args.conversation_id
first_id = query_args.first_id or None
try:
pagination = MessageService.pagination_by_first_id(
@@ -115,9 +117,7 @@ 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,21 +127,20 @@ 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(
@@ -178,11 +177,12 @@ 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,7 +190,6 @@ 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={
@@ -213,11 +212,12 @@ class AppGetFeedbacksApi(Resource):
feedbacks = MessageService.get_all_messages_feedbacks(
app_model, page=query_args.page, limit=query_args.limit, session=db.session()
)
return {"data": feedbacks}
return AppFeedbackListResponse(data=feedbacks).model_dump(mode="json")
@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,7 +237,6 @@ 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(
@@ -276,4 +275,4 @@ class MessageSuggestedApi(Resource):
logger.exception("internal server error.")
raise InternalServerError()
return {"result": "success", "data": questions}
return SimpleResultStringListResponse(result="success", data=questions).model_dump(mode="json")
+41 -63
View File
@@ -1,19 +1,24 @@
import logging
from collections.abc import Mapping
from datetime import datetime
from typing import Literal, override
from typing import Literal
from dateutil.parser import isoparse
from flask import request
from flask_restx import Resource, fields
from pydantic import BaseModel, Field, field_validator
from flask_restx import Resource
from pydantic import BaseModel, Field, field_validator, model_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, register_response_schema_models, register_schema_models
from controllers.common.schema import (
query_params_from_model,
query_params_from_request,
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 (
@@ -42,14 +47,13 @@ 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 SimpleAccount
from fields.member_fields import SimpleAccountResponse
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 to_timestamp
from libs.helper import dump_response, 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
@@ -105,29 +109,12 @@ 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)
outputs: dict = Field(default_factory=dict, validation_alias="outputs_dict")
error: str | None = None
total_steps: int | None = None
total_tokens: int | None = None
@@ -135,11 +122,33 @@ 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
@@ -171,7 +180,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: SimpleAccount | None = None
created_by_account: SimpleAccountResponse | None = None
created_by_end_user: SimpleEndUser | None = None
created_at: int | None = None
@@ -203,39 +212,6 @@ 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(
@@ -288,7 +264,7 @@ class WorkflowRunDetailApi(Resource):
)
if not workflow_run:
raise NotFound("Workflow run not found.")
return _serialize_workflow_run(workflow_run)
return dump_response(WorkflowRunResponse, workflow_run)
@service_api_ns.route("/workflows/run")
@@ -373,6 +349,7 @@ 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)
@@ -489,6 +466,7 @@ 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))
@@ -554,7 +532,7 @@ class WorkflowTaskStopApi(Resource):
# New graph engine command channel mechanism
GraphEngineManager(redis_client).send_stop_command(task_id)
return {"result": "success"}
return SimpleResultResponse(result="success").model_dump()
@service_api_ns.route("/workflows/logs")
@@ -587,7 +565,7 @@ class WorkflowAppLogApi(Resource):
Returns paginated workflow execution logs with filtering options.
"""
args = WorkflowLogQuery.model_validate(request.args.to_dict())
args = query_params_from_request(WorkflowLogQuery)
status = WorkflowExecutionStatus(args.status) if args.status else None
created_at_before = isoparse(args.created_at__before) if args.created_at__before else None
@@ -609,4 +587,4 @@ class WorkflowAppLogApi(Resource):
created_by_account=args.created_by_account,
)
return _serialize_workflow_log_pagination(workflow_app_log_pagination)
return dump_response(WorkflowAppLogPaginationResponse, workflow_app_log_pagination)
+13 -16
View File
@@ -681,10 +681,10 @@ class DatasetApi(DatasetApiResource):
dataset,
str(payload.permission) if payload.permission else None,
payload.partial_member_list,
db.session(),
session=db.session(),
)
dataset = DatasetService.update_dataset(session, dataset_id_str, update_data, current_user)
dataset = DatasetService.update_dataset(dataset_id_str, update_data, current_user, session=session)
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 dump_response(SimpleResultResponse, {"result": "success"}), 200
return SimpleResultResponse(result="success").model_dump(mode="json"), 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(db.session(), "knowledge", cid)
tags = TagService.get_tags("knowledge", cid, session=db.session())
return dump_response(KnowledgeTagListResponse, tags), 200
@service_api_ns.doc(
@@ -911,11 +911,8 @@ 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 = dump_response(
KnowledgeTagResponse,
{"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0},
)
return response, 200
response = KnowledgeTagResponse(id=tag.id, name=tag.name, type=tag.type, binding_count="0")
return response.model_dump(mode="json"), 200
@service_api_ns.doc(
summary="Update Knowledge Tag",
@@ -953,11 +950,8 @@ class DatasetTagsApi(DatasetApiResource):
binding_count = TagService.get_tag_binding_count(tag_id, db.session(), tag_type=TagType.KNOWLEDGE)
response = dump_response(
KnowledgeTagResponse,
{"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count},
)
return response, 200
response = KnowledgeTagResponse(id=tag.id, name=tag.name, type=tag.type, binding_count=str(binding_count))
return response.model_dump(mode="json"), 200
@service_api_ns.doc(
summary="Delete Knowledge Tag",
@@ -1088,5 +1082,8 @@ class DatasetTagsBindingStatusApi(DatasetApiResource):
tags = TagService.get_tags_by_target_id(
"knowledge", current_user.current_tenant_id, str(dataset_id), db.session()
)
tags_list = [{"id": tag.id, "name": tag.name} for tag in tags]
return dump_response(DatasetBoundTagListResponse, {"data": tags_list, "total": len(tags)}), 200
response = DatasetBoundTagListResponse(
data=[DatasetBoundTagResponse(id=tag.id, name=tag.name) for tag in tags],
total=len(tags),
)
return response.model_dump(mode="json"), 200
+105 -44
View File
@@ -6,14 +6,22 @@ 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, WithJsonSchema, field_validator, model_validator
from pydantic import (
BaseModel,
Field,
GetJsonSchemaHandler,
ValidationError,
WithJsonSchema,
field_validator,
model_validator,
)
from pydantic.json_schema import SkipJsonSchema
from sqlalchemy import desc, func, select
from werkzeug.exceptions import Forbidden, NotFound
@@ -26,9 +34,10 @@ from controllers.common.errors import (
TooManyFilesError,
UnsupportedFileTypeError,
)
from controllers.common.fields import BinaryFileResponse, UrlResponse
from controllers.common.fields import UrlResponse
from controllers.common.schema import (
query_params_from_model,
query_params_from_request,
register_enum_models,
register_response_schema_models,
register_schema_models,
@@ -56,6 +65,7 @@ from fields.document_fields import (
DocumentMetadataResponse,
DocumentResponse,
DocumentStatusListResponse,
normalize_enum,
)
from libs.helper import dump_response
from libs.login import current_user
@@ -281,38 +291,44 @@ class DocumentAndBatchResponse(ResponseModel):
batch: str
# Use SkipJsonSchema to support 3 metadata modes
class DocumentDetailResponse(ResponseModel):
id: str
position: int | None = None
data_source_type: str | None = None
data_source_info: dict[str, Any] | None = Field(default=None)
position: int | SkipJsonSchema[None] = None
data_source_type: str | SkipJsonSchema[None] = None
data_source_info: dict[str, Any] | SkipJsonSchema[None] = None
dataset_process_rule_id: str | 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
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
tokens: int | None = None
indexing_status: str | None = None
indexing_status: str | SkipJsonSchema[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
enabled: bool | SkipJsonSchema[None] = None
disabled_at: int | None = None
disabled_by: str | None = None
archived: bool | None = None
archived: bool | SkipJsonSchema[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
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
display_status: str | None = None
doc_form: str | None = None
doc_form: str | SkipJsonSchema[None] = None
doc_language: str | None = None
summary_index_status: str | None = None
need_summary: bool | 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)
register_enum_models(service_api_ns, RetrievalMethod)
@@ -332,7 +348,6 @@ register_schema_models(
)
register_response_schema_models(
service_api_ns,
BinaryFileResponse,
UrlResponse,
DocumentResponse,
DocumentAndBatchResponse,
@@ -342,13 +357,13 @@ register_response_schema_models(
)
def _create_document_by_text(tenant_id: str, dataset_id: UUID) -> tuple[Mapping[str, object], int]:
def _create_document_by_text(tenant_id: str, dataset_id: UUID) -> tuple[Document, str]:
"""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 = str(tenant_id)
tenant_id_str = tenant_id
dataset = db.session.scalar(
select(Dataset).where(Dataset.tenant_id == tenant_id_str, Dataset.id == dataset_id_str).limit(1)
)
@@ -407,10 +422,10 @@ def _create_document_by_text(tenant_id: str, dataset_id: UUID) -> tuple[Mapping[
raise ProviderNotInitializeError(ex.description)
document = documents[0]
return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200
return document, batch
def _update_document_by_text(tenant_id: str, dataset_id: UUID, document_id: UUID) -> tuple[Mapping[str, object], int]:
def _update_document_by_text(tenant_id: str, dataset_id: UUID, document_id: UUID) -> tuple[Document, str]:
"""Update a document from text for both canonical and legacy routes."""
payload = DocumentTextUpdate.model_validate(service_api_ns.payload or {})
dataset = db.session.scalar(
@@ -467,7 +482,7 @@ def _update_document_by_text(tenant_id: str, dataset_id: UUID, document_id: UUID
raise ProviderNotInitializeError(ex.description)
document = documents[0]
return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200
return document, batch
@service_api_ns.route("/datasets/<uuid:dataset_id>/document/create-by-text")
@@ -511,7 +526,8 @@ class DocumentAddByTextApi(DatasetApiResource):
@cloud_edition_billing_rate_limit_check("knowledge", "dataset")
def post(self, tenant_id: str, dataset_id: UUID):
"""Create document by text."""
return _create_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id)
document, batch = _create_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id)
return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200
@service_api_ns.route("/datasets/<uuid:dataset_id>/document/create_by_text")
@@ -543,7 +559,8 @@ 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."""
return _create_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id)
document, batch = _create_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id)
return dump_response(DocumentAndBatchResponse, {"document": document, "batch": batch}), 200
@service_api_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/update-by-text")
@@ -587,7 +604,8 @@ 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."""
return _update_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
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
@service_api_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/update_by_text")
@@ -618,7 +636,8 @@ 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."""
return _update_document_by_text(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
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
@service_api_ns.route(
@@ -767,10 +786,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[Mapping[str, object], int]:
def _update_document_by_file(tenant_id: str, dataset_id: UUID, document_id: UUID) -> tuple[Document, str]:
"""Update a document from an uploaded file for canonical and deprecated routes."""
dataset_id_str = str(dataset_id)
tenant_id_str = str(tenant_id)
tenant_id_str = tenant_id
dataset = db.session.scalar(
select(Dataset).where(Dataset.tenant_id == tenant_id_str, Dataset.id == dataset_id_str).limit(1)
)
@@ -841,7 +860,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 dump_response(DocumentAndBatchResponse, {"document": document, "batch": document.batch}), 200
return document, document.batch
@service_api_ns.route(
@@ -895,7 +914,8 @@ 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."""
return _update_document_by_file(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
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
@service_api_ns.route("/datasets/<uuid:dataset_id>/documents")
@@ -928,7 +948,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 = DocumentListQuery.model_validate(request.args.to_dict())
query_params = query_params_from_request(DocumentListQuery)
dataset = db.session.scalar(
select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id_str).limit(1)
)
@@ -1021,6 +1041,7 @@ class DocumentBatchDownloadZipApi(DatasetApiResource):
)
cleanup = stack.pop_all()
response.call_on_close(cleanup.close)
# response-contract:ignore binary send_file response
return response
@@ -1149,7 +1170,9 @@ class DocumentDownloadApi(DatasetApiResource):
if document.tenant_id != str(tenant_id):
raise Forbidden("No permission.")
return {"url": DocumentService.get_document_download_url(document, db.session())}
return UrlResponse(url=DocumentService.get_document_download_url(document, db.session())).model_dump(
mode="json"
)
@service_api_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")
@@ -1176,8 +1199,13 @@ 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."})
@service_api_ns.doc(params=query_params_from_model(DocumentGetQuery))
@service_api_ns.doc(
params={
"dataset_id": "Knowledge base ID.",
"document_id": "Document ID.",
**query_params_from_model(DocumentGetQuery),
}
)
@service_api_ns.doc(
responses={
200: "Document retrieved successfully",
@@ -1205,9 +1233,14 @@ class DocumentApi(DatasetApiResource):
if document.tenant_id != str(tenant_id):
raise Forbidden("No permission.")
metadata = request.args.get("metadata", "all")
if metadata not in self.METADATA_CHOICES:
raise InvalidMetadataError(f"Invalid metadata value: {metadata}")
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
# Calculate summary_index_status if needed
summary_index_status = None
@@ -1221,8 +1254,10 @@ 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
@@ -1295,8 +1330,33 @@ class DocumentApi(DatasetApiResource):
"need_summary": document.need_summary if document.need_summary is not None else False,
}
return response
return DocumentDetailResponse.model_validate(response).model_dump(
mode="json",
include=response_include,
exclude=response_exclude,
)
@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)
@@ -1314,7 +1374,8 @@ 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."""
return _update_document_by_file(tenant_id=tenant_id, dataset_id=dataset_id, document_id=document_id)
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
@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(db.session(), dataset_id_str, metadata_args)
metadata = MetadataService.create_metadata(dataset_id_str, metadata_args, session=db.session())
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(db.session(), dataset)
metadata = MetadataService.get_dataset_metadatas(dataset, session=db.session())
return dump_response(DatasetMetadataListResponse, metadata), 200
@@ -159,7 +159,9 @@ class DatasetMetadataServiceApi(DatasetApiResource):
raise NotFound("Dataset not found.")
DatasetService.check_dataset_permission(dataset, current_user, db.session())
metadata = MetadataService.update_metadata_name(db.session(), dataset_id_str, metadata_id_str, payload.name)
metadata = MetadataService.update_metadata_name(
dataset_id_str, metadata_id_str, payload.name, session=db.session()
)
return dump_response(DatasetMetadataResponse, metadata), 200
@service_api_ns.doc(
@@ -194,7 +196,7 @@ class DatasetMetadataServiceApi(DatasetApiResource):
raise NotFound("Dataset not found.")
DatasetService.check_dataset_permission(dataset, current_user, db.session())
MetadataService.delete_metadata(db.session(), dataset_id_str, metadata_id_str)
MetadataService.delete_metadata(dataset_id_str, metadata_id_str, session=db.session())
return "", 204
@@ -264,9 +266,9 @@ class DatasetMetadataBuiltInFieldActionServiceApi(DatasetApiResource):
match action:
case "enable":
MetadataService.enable_built_in_field(db.session(), dataset)
MetadataService.enable_built_in_field(dataset, session=db.session())
case "disable":
MetadataService.disable_built_in_field(db.session(), dataset)
MetadataService.disable_built_in_field(dataset, session=db.session())
return dump_response(DatasetMetadataActionResponse, {"result": "success"}), 200
@@ -310,6 +312,6 @@ class DocumentMetadataEditServiceApi(DatasetApiResource):
metadata_args = MetadataOperationData.model_validate(service_api_ns.payload or {})
MetadataService.update_documents_metadata(db.session(), dataset, metadata_args)
MetadataService.update_documents_metadata(dataset, metadata_args, session=db.session())
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 = db.session.scalar(stmt)
dataset = 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(db.session())
rag_pipeline_service = RagPipelineService(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(
+4 -2
View File
@@ -180,7 +180,9 @@ 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(db.session(), app_id=app_id)
require_permission_check = WebAppAuthService.is_app_require_permission_check(
app_id=app_id, session=db.session()
)
if not require_permission_check:
return {"result": True}
@@ -201,6 +203,6 @@ class AppWebAuthPermission(Resource):
return {"result": True}
res = True
if WebAppAuthService.is_app_require_permission_check(db.session(), app_id=app_id):
if WebAppAuthService.is_app_require_permission_check(app_id=app_id, session=db.session()):
res = EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp(str(user_id), app_id)
return {"result": res}
+12 -15
View File
@@ -1,13 +1,11 @@
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,
@@ -23,8 +21,9 @@ 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 uuid_value
from libs.helper import dump_response, uuid_value
from models.model import App, EndUser
from services.app_ref_service import AppRefService
from services.audio_service import AudioService
@@ -38,6 +37,10 @@ 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
@@ -48,18 +51,13 @@ class TextToAudioPayload(TextToAudioPayloadBase):
register_schema_models(web_ns, TextToAudioPayload)
register_response_schema_models(web_ns, AudioBinaryResponse, AudioTranscriptResponse)
register_response_schema_models(web_ns, AudioToTextResponse)
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(
@@ -73,7 +71,7 @@ class AudioApi(WebApiResource):
500: "Internal Server Error",
}
)
@web_ns.response(200, "Success", web_ns.models[AudioTranscriptResponse.__name__])
@web_ns.response(200, "Success", web_ns.models[AudioToTextResponse.__name__])
def post(self, app_model: App, end_user: EndUser):
"""Convert audio to text"""
file = request.files["file"]
@@ -81,7 +79,7 @@ class AudioApi(WebApiResource):
try:
response = AudioService.transcript_asr(app_model=app_model, file=file, end_user=end_user.external_user_id)
return response
return dump_response(AudioToTextResponse, response)
except services.errors.app_model_config.AppModelConfigBrokenError:
logger.exception("App model config broken.")
raise AppUnavailableError()
@@ -122,7 +120,8 @@ class TextApi(WebApiResource):
500: "Internal Server Error",
}
)
@web_ns.response(200, "Success", web_ns.models[AudioBinaryResponse.__name__])
# response-contract:ignore provider audio bytes; TODO: model binary audio response if shape is standardized.
@web_ns.response(200, "Success")
def post(self, app_model: App, end_user: EndUser):
"""Convert text to audio"""
try:
@@ -139,7 +138,7 @@ class TextApi(WebApiResource):
message_id,
end_user_id=end_user.id,
)
response = AudioService.transcript_tts(
return AudioService.transcript_tts(
app_model=app_model,
session=db.session(),
text=text,
@@ -147,8 +146,6 @@ 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()
+4 -2
View File
@@ -133,6 +133,7 @@ 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.")
@@ -185,7 +186,7 @@ class CompletionStopApi(WebApiResource):
app_mode=AppMode.value_of(app_model.mode),
)
return {"result": "success"}, 200
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
@web_ns.route("/chat-messages")
@@ -235,6 +236,7 @@ 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.")
@@ -290,4 +292,4 @@ class ChatStopApi(WebApiResource):
app_mode=app_mode,
)
return {"result": "success"}, 200
return SimpleResultResponse(result="success").model_dump(mode="json"), 200
+2 -2
View File
@@ -13,6 +13,7 @@ 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
@@ -84,5 +85,4 @@ class FileApi(WebApiResource):
except services.errors.file.UnsupportedFileTypeError:
raise UnsupportedFileTypeError()
response = FileResponse.model_validate(upload_file, from_attributes=True)
return response.model_dump(mode="json"), 201
return dump_response(FileResponse, upload_file), 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(db.session(), request_email)
account = AccountService.get_account_by_email_with_case_fallback(request_email, session=db.session())
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(db.session(), email)
account = AccountService.get_account_by_email_with_case_fallback(email, session=db.session())
if account:
account = db.session.merge(account)
@@ -29,6 +29,7 @@ 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 (
@@ -141,8 +142,7 @@ def _upload_local_file(context):
except services.errors.file.BlockedFileExtensionError as exc:
raise BlockedFileExtensionError() from exc
response = FileResponse.model_validate(upload_file, from_attributes=True)
return upload_file.id, response
return upload_file.id, dump_response(FileResponse, upload_file)
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
return upload_file.id, response.model_dump(mode="json")
@web_ns.route("/human-input-forms/files")
@@ -209,4 +209,5 @@ class HumanInputFileUploadApi(Resource):
file_id, response = _upload_local_file(context=context)
upload_service.record_upload_file(context=context, file_id=file_id)
return response.model_dump(mode="json"), 201
# response-contract:ignore pre-dumped response. See above
return response, 201
+109 -58
View File
@@ -2,14 +2,12 @@
Web App Human Input Form APIs.
"""
import json
import logging
from collections.abc import Sequence
from typing import Any, NotRequired, TypedDict
from typing import Self
from flask import Response, request
from flask import 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
@@ -20,35 +18,58 @@ 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 serialize_app_site_payload
from core.workflow.nodes.human_input.entities import FormInputConfig
from controllers.web.site import WebAppSiteResponse
from core.workflow.nodes.human_input.entities import FormInputConfig, UserActionConfig
from extensions.ext_database import db
from libs.helper import RateLimiter, extract_remote_ip, to_timestamp
from fields.base import ResponseModel
from libs.helper import RateLimiter, dump_response, 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(BaseModel):
class HumanInputUploadTokenResponse(ResponseModel):
upload_token: str
expires_at: int
class HumanInputFormDefinitionResponse(BaseModel):
form_content: Any
inputs: Any
class HumanInputFormDefinitionResponse(ResponseModel):
form_content: str
inputs: list[FormInputConfig]
resolved_default_values: dict[str, str]
user_actions: Any
user_actions: list[UserActionConfig]
expiration_time: int
site: dict[str, Any] | None = Field(default=None)
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,
)
class HumanInputFormSubmitResponse(BaseModel):
model_config = ConfigDict(extra="forbid")
class HumanInputFormSubmitResponse(ResponseModel):
pass
register_schema_models(web_ns, HumanInputFormSubmitPayload)
@@ -86,40 +107,26 @@ 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.response(200, "Success", web_ns.models[HumanInputUploadTokenResponse.__name__])
@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__],
)
def post(self, form_token: str):
"""
Issue an upload token for a human input form.
@@ -136,11 +143,9 @@ class HumanInputFormUploadTokenApi(Resource):
except FormNotFoundError:
raise NotFoundError("Form not found")
response = HumanInputUploadTokenResponse(
upload_token=token.upload_token,
expires_at=to_timestamp(token.expires_at),
)
return response.model_dump(mode="json"), 200
return HumanInputUploadTokenResponse(
upload_token=token.upload_token, expires_at=to_timestamp(token.expires_at)
).model_dump(mode="json"), 200
@web_ns.route("/form/human_input/<string:form_token>")
@@ -150,7 +155,23 @@ 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.response(200, "Success", web_ns.models[HumanInputFormDefinitionResponse.__name__])
@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__],
)
def get(self, form_token: str):
"""
Get human input form definition by token.
@@ -172,17 +193,47 @@ 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 _jsonify_form_definition(
form,
inputs=inputs,
site_payload=serialize_app_site_payload(app_model, site, None),
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,
),
),
)
# 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.
@@ -225,7 +276,7 @@ class HumanInputFormApi(Resource):
except FormNotFoundError:
raise NotFoundError("Form not found")
return {}, 200
return HumanInputFormSubmitResponse().model_dump(mode="json"), 200
def _get_app_site_from_form(form: Form) -> tuple[App, Site]:
@@ -238,7 +289,7 @@ def _get_app_site_from_form(form: Form) -> tuple[App, Site]:
if site is None:
raise Forbidden()
if app_model.tenant and app_model.tenant.status == TenantStatus.ARCHIVE:
if app_model.tenant is None or 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(
db.session(), app_id=app_id
app_id=app_id, session=db.session()
)
user_logged_in = False
+1
View File
@@ -188,6 +188,7 @@ 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(db.session(), app_code=app_code)
app_auth_type = WebAppAuthService.get_app_auth_type(app_code=app_code, session=db.session())
if app_auth_type != WebAppAuthType.PUBLIC:
if not enterprise_user_decoded:
raise WebAppAuthRequiredError()
+4 -6
View File
@@ -65,11 +65,10 @@ class RemoteFileInfoApi(WebApiResource):
# failed back to get method
resp = remote_fetcher.make_request("GET", decoded_url, timeout=3)
resp.raise_for_status()
info = RemoteFileInfo(
return RemoteFileInfo(
file_type=resp.headers.get("Content-Type", "application/octet-stream"),
file_length=int(resp.headers.get("Content-Length", -1)),
)
return info.model_dump(mode="json")
).model_dump(mode="json")
@web_ns.route("/remote-files/upload")
@@ -141,7 +140,7 @@ class RemoteFileUploadApi(WebApiResource):
except services.errors.file.UnsupportedFileTypeError:
raise UnsupportedFileTypeError
payload1 = FileWithSignedUrl(
return FileWithSignedUrl(
id=upload_file.id,
name=upload_file.name,
size=upload_file.size,
@@ -150,5 +149,4 @@ class RemoteFileUploadApi(WebApiResource):
mime_type=upload_file.mime_type,
created_by=upload_file.created_by,
created_at=int(upload_file.created_at.timestamp()),
)
return payload1.model_dump(mode="json"), 201
).model_dump(mode="json"), 201
+5 -6
View File
@@ -44,14 +44,12 @@ class SavedMessageListApi(WebApiResource):
query = SavedMessageListQuery.model_validate(raw_args)
pagination = SavedMessageService.pagination_by_last_id(
db.session(), app_model, end_user, query.last_id, query.limit
app_model, end_user, query.last_id, query.limit, session=db.session()
)
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")
@@ -80,7 +78,7 @@ class SavedMessageListApi(WebApiResource):
payload = SavedMessageCreatePayload.model_validate(web_ns.payload or {})
try:
SavedMessageService.save(db.session(), app_model, end_user, payload.message_id)
SavedMessageService.save(app_model, end_user, payload.message_id, session=db.session())
except MessageNotExistsError:
raise NotFound("Message Not Exists.")
@@ -102,12 +100,13 @@ 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(db.session(), app_model, end_user, message_id_str)
SavedMessageService.delete(app_model, end_user, message_id_str, session=db.session())
return "", 204
+100 -132
View File
@@ -1,7 +1,6 @@
from typing import Any, cast
from typing import Any, Self
from flask_restx import fields, marshal, marshal_with
from pydantic import Field
from pydantic import AliasChoices, Field, computed_field
from sqlalchemy import select
from werkzeug.exceptions import Forbidden
@@ -11,30 +10,19 @@ 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 AppIconUrlField
from models.account import TenantStatus
from libs.helper import build_icon_url
from models.account import Tenant, TenantStatus
from models.model import App, EndUser, Site
from services.feature_service import FeatureModel, FeatureService
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
class WebSiteResponse(ResponseModel):
title: str
chat_color_theme: str | None = None
chat_color_theme_inverted: bool | None = None
chat_color_theme_inverted: bool
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
@@ -45,65 +33,98 @@ class AppSiteResponse(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 AppSiteInfoResponse(ResponseModel):
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):
app_id: str
end_user_id: str | None = None
enable_site: bool
site: AppSiteResponse
model_config_: AppSiteModelConfigResponse | None = Field(default=None, alias="model_config")
plan: str | None = None
site: WebSiteResponse
model_config_: WebModelConfigResponse | None = Field(
default=None, validation_alias="model_config", serialization_alias="model_config"
)
plan: str
can_replace_logo: bool
custom_config: dict[str, Any] | None = Field(default=None)
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,
)
register_response_schema_models(web_ns, AppSiteInfoResponse)
register_response_schema_models(
web_ns, WebSiteResponse, WebModelConfigResponse, WebAppCustomConfigResponse, WebAppSiteResponse
)
@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(
@@ -116,79 +137,26 @@ class AppSiteApi(WebApiResource):
500: "Internal Server Error",
}
)
@web_ns.response(200, "Success", web_ns.models[AppSiteInfoResponse.__name__])
@marshal_with(app_fields)
@web_ns.response(200, "Success", web_ns.models[WebAppSiteResponse.__name__])
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 not site:
if site is None:
raise Forbidden()
if app_model.tenant and app_model.tenant.status == TenantStatus.ARCHIVE:
tenant = app_model.tenant
if tenant is None or tenant.status == TenantStatus.ARCHIVE:
raise Forbidden()
features = FeatureService.get_features(app_model.tenant_id, exclude_vector_space=True)
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))
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")
+2 -1
View File
@@ -76,6 +76,7 @@ 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)
@@ -129,4 +130,4 @@ class WorkflowTaskStopApi(WebApiResource):
# New graph engine command channel mechanism
GraphEngineManager(redis_client).send_stop_command(task_id)
return {"result": "success"}
return SimpleResultResponse(result="success").model_dump(mode="json")
+3 -1
View File
@@ -129,7 +129,9 @@ def _validate_user_accessibility(
if not webapp_settings:
raise WebAppAuthRequiredError("Web app settings not found.")
if WebAppAuthService.is_app_require_permission_check(db.session(), access_mode=webapp_settings.access_mode):
if WebAppAuthService.is_app_require_permission_check(
access_mode=webapp_settings.access_mode, session=db.session()
):
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,6 +48,7 @@ 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
@@ -269,6 +270,7 @@ 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.
@@ -298,6 +300,7 @@ 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(
@@ -492,6 +495,7 @@ 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.
@@ -539,12 +543,14 @@ 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,
)
)
@@ -565,6 +571,7 @@ 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,
},
)
@@ -604,6 +611,7 @@ 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.
@@ -663,6 +671,7 @@ 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,6 +44,7 @@ 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
@@ -78,6 +79,7 @@ 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,
@@ -95,6 +97,7 @@ 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):
@@ -241,6 +244,7 @@ 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
+81 -89
View File
@@ -1,14 +1,11 @@
"""Agent App generator: orchestrate Agent App chat and finalize executions.
The primary mode mirrors the agent_chat generator (conversation + message +
Agent App turns mirror 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.
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.
LLM/ReAct loop. Build-chat finalization uses this same streamed path and only
changes the runtime exit policy carried to the backend.
"""
from __future__ import annotations
@@ -41,13 +38,16 @@ 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,12 +64,68 @@ 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 raw request file references to the backend user prompt."""
if not prompt_file_mappings:
"""Append labeled, prompt-safe file locators to the backend user prompt."""
prompt_files = _prompt_file_locators(prompt_file_mappings)
if not prompt_files:
return query
return f"{query}\n{json.dumps(list(prompt_file_mappings), ensure_ascii=False)}"
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
class AgentAppGenerator(MessageBasedAppGenerator):
@@ -120,6 +176,7 @@ 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()),
@@ -149,6 +206,7 @@ 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)
@@ -187,86 +245,6 @@ 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,
*,
@@ -355,6 +333,7 @@ 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,
@@ -476,6 +455,7 @@ 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
@@ -492,6 +472,18 @@ 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)
+328 -171
View File
@@ -1,15 +1,10 @@
"""Agent App runner: drive Agent backend turns for both chat and finalize flows.
"""Agent App runner: drive Agent backend turns for chat and finalization flows.
Unlike the legacy ``AgentChatAppRunner`` (which runs an in-process ReAct loop),
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.
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.
"""
from __future__ import annotations
@@ -26,16 +21,18 @@ 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 configs import dify_config
from clients.agent_backend.session_cleanup import AgentBackendSessionCleanupPayload
from core.app.apps.agent_app.runtime_request_builder import (
AgentAppRuntimeBuildContext,
AgentAppRuntimeRequest,
@@ -48,17 +45,31 @@ 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 DifyRunContext
from core.app.entities.queue_entities import QueueAgentThoughtEvent, QueueLLMChunkEvent, QueueMessageEndEvent
from core.app.entities.app_invoke_entities import AgentRuntimeExitIntent, DifyRunContext
from core.app.entities.queue_entities import (
QueueAgentMessageEvent,
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__)
@@ -69,6 +80,22 @@ 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:
@@ -135,6 +162,25 @@ 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,
@@ -159,7 +205,7 @@ def publish_message_end(
class _TextDeltaDebouncer:
"""Batch assistant text deltas on stream-event boundaries for final SSE output."""
"""Batch independent model text deltas before agent-message SSE output."""
def __init__(self, *, debounce_seconds: float) -> None:
self._debounce_seconds = debounce_seconds
@@ -192,7 +238,13 @@ class _TextDeltaDebouncer:
class _AgentProcessRecorder:
"""Persist Agent v2 thinking/tool process events through the legacy thought model."""
"""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.
"""
def __init__(
self,
@@ -206,6 +258,7 @@ 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]] = {}
@@ -261,6 +314,41 @@ 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):
@@ -281,6 +369,7 @@ 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)
@@ -289,6 +378,7 @@ 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")
@@ -305,8 +395,10 @@ 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)
@@ -322,8 +414,10 @@ 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")
@@ -332,6 +426,7 @@ 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)
@@ -342,12 +437,15 @@ 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:
self._tool_by_index[index] = thought_id
if index >= 0:
self._tool_by_index[index] = thought_id
if tool_call_id:
self._tool_by_call_id[tool_call_id] = thought_id
if tool_name:
@@ -363,11 +461,24 @@ 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, tool: str | None = None, tool_input: str | None = None
self,
*,
thought: str | None = None,
answer: str | None = None,
tool: str | None = None,
tool_input: str | None = None,
) -> str:
row = MessageAgentThought(
message_id=self._message_id,
@@ -384,7 +495,7 @@ class _AgentProcessRecorder:
message_unit_price=Decimal(0),
message_price_unit=Decimal("0.001"),
message_files="",
answer="",
answer=answer or "",
answer_token=0,
answer_unit_price=Decimal(0),
answer_price_unit=Decimal("0.001"),
@@ -414,6 +525,7 @@ 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:
@@ -430,6 +542,8 @@ 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(
@@ -448,7 +562,12 @@ def _event_index(data: dict[str, Any]) -> int:
def _string_or_none(value: Any) -> str | None:
return value if isinstance(value, str) and value else None
if not isinstance(value, str):
return None
normalized = value.strip()
if not normalized or normalized.lower() in {"none", "null"}:
return None
return normalized
def _json_or_text(value: Any) -> str | None:
@@ -468,6 +587,18 @@ 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."""
@@ -500,7 +631,9 @@ 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,
@@ -522,10 +655,11 @@ 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, streamed_answer = self._consume_stream(
terminal, process_recorder = self._consume_stream(
create_response.run_id,
dify_context=dify_context,
message_id=message_id,
@@ -535,6 +669,9 @@ 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(
@@ -552,73 +689,53 @@ class AgentAppRunner:
return
if not isinstance(terminal, AgentBackendRunSucceededInternalEvent):
error = getattr(terminal, "error", None) or "Agent backend run did not complete successfully."
raise AgentBackendError(str(error))
if isinstance(terminal, AgentBackendRunFailedInternalEvent):
raise _agent_backend_failure_to_exception(terminal)
raise AgentBackendError("Agent backend run did not complete successfully.")
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))
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)
def _build_session_scope(
self,
@@ -654,6 +771,7 @@ 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 = (
@@ -673,6 +791,7 @@ class AgentAppRunner:
idempotency_key=idempotency_key,
session_snapshot=session_snapshot,
deferred_tool_results=deferred_tool_results,
suspend_on_exit=suspend_on_exit,
)
)
@@ -775,46 +894,61 @@ class AgentAppRunner:
model_name: str,
query: str | None,
):
"""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.
"""
"""Consume backend events while preserving raw recorder granularity."""
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 flush_pending_text() -> None:
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:
pending_text = text_delta_debouncer.flush()
if pending_text:
publish_text_delta(
queue_manager=queue_manager,
model_name=model_name,
delta=pending_text,
user_query=query,
)
persist_answer_text(pending_text)
for public_event in self._agent_backend_client.stream_events(run_id):
if queue_manager.is_stopped():
flush_pending_text()
flush_pending_agent_message_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_text()
flush_pending_agent_message_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:
@@ -826,26 +960,15 @@ 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_text()
flush_pending_agent_message_text()
terminal = internal_event
break
if terminal is not None:
break
flush_pending_text()
return terminal, "".join(streamed_answer_parts)
flush_pending_agent_message_text()
return terminal, process_recorder
def _cancel_run(self, run_id: str) -> None:
try:
@@ -867,42 +990,15 @@ class AgentAppRunner:
model_name: str,
answer: str,
query: str | None,
streamed_answer: str,
usage: LLMUsage | None,
) -> None:
"""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(
"""Finish a successful turn from the backend terminal output."""
publish_text_answer(
queue_manager=queue_manager,
model_name=model_name,
answer=answer,
user_query=query,
usage=usage,
user_query=query,
)
def _save_session(
@@ -914,7 +1010,7 @@ class AgentAppRunner:
runtime_layer_specs: Any,
pending_form_id: str | None = None,
pending_tool_call_id: str | None = None,
) -> None:
) -> bool:
try:
self._session_store.save_active_snapshot(
scope=scope,
@@ -924,6 +1020,7 @@ 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: "
@@ -934,9 +1031,91 @@ 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 _extract_answer(output: JsonValue) -> str:
def _terminal_output_to_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
@@ -954,27 +1133,5 @@ 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,6 +74,7 @@ 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)
@@ -163,6 +164,7 @@ 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,
@@ -173,6 +175,7 @@ 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,
)
@@ -124,6 +124,41 @@ class AgentAppRuntimeSessionStore:
runtime_layer_specs=_deserialize_runtime_layer_specs(row.composition_layer_specs),
)
def list_active_sessions_for_conversation(
self, *, tenant_id: str, app_id: str, conversation_id: str
) -> list[StoredAgentAppSession]:
"""List all ACTIVE conversation-owned sessions for lifecycle cleanup."""
stmt = (
select(AgentRuntimeSession)
.where(
AgentRuntimeSession.owner_type == AgentRuntimeSessionOwnerType.CONVERSATION,
AgentRuntimeSession.tenant_id == tenant_id,
AgentRuntimeSession.app_id == app_id,
AgentRuntimeSession.conversation_id == conversation_id,
AgentRuntimeSession.status == AgentRuntimeSessionStatus.ACTIVE,
)
.order_by(AgentRuntimeSession.updated_at.desc())
)
with session_factory.create_session() as session:
rows = session.scalars(stmt).all()
return [
StoredAgentAppSession(
scope=AgentAppSessionScope(
tenant_id=row.tenant_id,
app_id=row.app_id,
conversation_id=row.conversation_id or "",
agent_id=row.agent_id,
agent_config_snapshot_id=row.agent_config_snapshot_id,
),
session_snapshot=CompositorSessionSnapshot.model_validate_json(row.session_snapshot),
backend_run_id=row.backend_run_id,
runtime_layer_specs=_deserialize_runtime_layer_specs(row.composition_layer_specs),
pending_form_id=row.pending_form_id,
pending_tool_call_id=row.pending_tool_call_id,
)
for row in rows
]
def save_active_snapshot(
self,
*,
@@ -134,6 +169,14 @@ class AgentAppRuntimeSessionStore:
pending_form_id: str | None = None,
pending_tool_call_id: str | None = None,
) -> None:
"""Persist the current conversation snapshot and enforce one ACTIVE row.
Agent App chat treats one conversation as one resumable runtime shell.
Saving the latest snapshot therefore upserts the scoped row back to
ACTIVE and retires any other ACTIVE conversation-owned rows for the
same ``tenant_id + app_id + conversation_id`` so later lookups see a
single active session.
"""
if snapshot is None:
return
snapshot_json = snapshot.model_dump_json()
@@ -205,6 +205,7 @@ class AgentChatAppGenerator(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,
@@ -8,7 +8,7 @@ from pydantic import JsonValue
from core.app.entities.app_invoke_entities import InvokeFrom
from core.app.entities.task_entities import AppBlockingResponse, AppStreamResponse
from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
from graphon.model_runtime.errors.invoke import InvokeError
from graphon.model_runtime.errors.invoke import InvokeError, InvokeRateLimitError
logger = logging.getLogger(__name__)
@@ -127,6 +127,7 @@ class AppGenerateResponseConverter[TBlockingResponse: AppBlockingResponse](ABC):
},
ModelCurrentlyNotSupportError: {"code": "model_currently_not_support", "status": 400},
InvokeError: {"code": "completion_request_error", "status": 400},
InvokeRateLimitError: {"code": "rate_limit_error", "status": 429},
}
# Determine the response based on the type of exception

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