opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-05-10 05:56:31 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(api): migrate console.app.workflow etc. to BaseModel (#35967)

Browse Source
This commit is contained in:
chariri 2026-05-09 17:34:15 +09:00 committed by GitHub
parent 861f73267c
commit 4a56763d2f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 941 additions and 498 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
api/AGENTS.md
View File

@ -193,6 +193,10 @@ Before opening a PR / submitting:
- Controllers: parse input via Pydantic, invoke services, return serialised responses; no business logic.
- Services: coordinate repositories, providers, background tasks; keep side effects explicit.
- Document non-obvious behaviour with concise docstrings and comments.
- For Flask-RESTX controller request, query, and response schemas, follow `controllers/API_SCHEMA_GUIDE.md`.
In short: use Pydantic models, document GET query params with `query_params_from_model(...)`, register response
DTOs with `register_response_schema_models(...)`, serialize with `ResponseModel.model_validate(...).model_dump(...)`,
and avoid adding new legacy `ns.model(...)`, `@marshal_with(...)`, or GET `@ns.expect(...)` patterns.
### Miscellaneous

193
api/controllers/API_SCHEMA_GUIDE.md Normal file
View File

@ -0,0 +1,193 @@
# API Schema Guide
This guide describes the expected Flask-RESTX + Pydantic pattern for controller request payloads, query
parameters, response schemas, and Swagger documentation.
## Principles
- Use Pydantic `BaseModel` for request bodies and query parameters.
- Use `fields.base.ResponseModel` for response DTOs.
- Keep runtime validation and Swagger documentation wired to the same Pydantic model.
- Prefer explicit validation and serialization in controller methods over Flask-RESTX marshalling.
- Do not add new Flask-RESTX `fields.*` dictionaries, `Namespace.model(...)` exports, or `@marshal_with(...)` for migrated or new endpoints.
- Do not use `@ns.expect(...)` for GET query parameters. Flask-RESTX documents that as a request body.
## Naming
- Request body models: use a `Payload` suffix.
- Example: `WorkflowRunPayload`, `DatasourceVariablesPayload`.
- Query parameter models: use a `Query` suffix.
- Example: `WorkflowRunListQuery`, `MessageListQuery`.
- Response models: use a `Response` suffix and inherit from `ResponseModel`.
- Example: `WorkflowRunDetailResponse`, `WorkflowRunNodeExecutionListResponse`.
- Use `ListResponse` or `PaginationResponse` for wrapper responses.
- Example: `WorkflowRunNodeExecutionListResponse`, `WorkflowRunPaginationResponse`.
- Keep these models near the controller when they are endpoint-specific. Move them to `fields/*_fields.py` only when shared by multiple controllers.
## Registering Models For Swagger
Use helpers from `controllers.common.schema`.
```python
from controllers.common.schema import (
query_params_from_model,
register_response_schema_models,
register_schema_models,
)
```
Register request payload and query models with `register_schema_models(...)`:
```python
register_schema_models(
console_ns,
WorkflowRunPayload,
WorkflowRunListQuery,
)
```
Register response models with `register_response_schema_models(...)`:
```python
register_response_schema_models(
console_ns,
WorkflowRunDetailResponse,
WorkflowRunPaginationResponse,
)
```
Response models are registered in Pydantic serialization mode. This matters when a response model uses
`validation_alias` to read internal object attributes but emits public API field names. For example, a response model
can validate from `inputs_dict` while documenting and serializing `inputs`.
## Request Bodies
For non-GET request bodies:
1. Define a Pydantic `Payload` model.
2. Register it with `register_schema_models(...)`.
3. Use `@ns.expect(ns.models[Payload.__name__])` for Swagger documentation.
4. Validate from `ns.payload or {}` inside the controller.
```python
class DraftWorkflowNodeRunPayload(BaseModel):
inputs: dict[str, Any]
query: str = ""
register_schema_models(console_ns, DraftWorkflowNodeRunPayload)
@console_ns.expect(console_ns.models[DraftWorkflowNodeRunPayload.__name__])
def post(self, app_model: App, node_id: str):
payload = DraftWorkflowNodeRunPayload.model_validate(console_ns.payload or {})
result = service.run(..., inputs=payload.inputs, query=payload.query)
return WorkflowRunNodeExecutionResponse.model_validate(result, from_attributes=True).model_dump(mode="json")
```
## Query Parameters
For GET query parameters:
1. Define a Pydantic `Query` model.
2. Register it with `register_schema_models(...)` if it is referenced elsewhere in docs, or only use
`query_params_from_model(...)` if a body schema is not needed.
3. Use `@ns.doc(params=query_params_from_model(QueryModel))`.
4. Validate from `request.args.to_dict(flat=True)` or an explicit dict when type coercion is needed.
```python
class WorkflowRunListQuery(BaseModel):
last_id: str | None = Field(default=None, description="Last run ID for pagination")
limit: int = Field(default=20, ge=1, le=100, description="Number of items per page (1-100)")
@console_ns.doc(params=query_params_from_model(WorkflowRunListQuery))
def get(self, app_model: App):
query = WorkflowRunListQuery.model_validate(request.args.to_dict(flat=True))
result = service.list(..., limit=query.limit, last_id=query.last_id)
return WorkflowRunPaginationResponse.model_validate(result, from_attributes=True).model_dump(mode="json")
```
Do not do this for GET query parameters:
```python
@console_ns.expect(console_ns.models[WorkflowRunListQuery.__name__])
def get(...):
...
```
That documents a GET request body and is not the expected contract.
## Responses
Response models should inherit from `ResponseModel`:
```python
class WorkflowRunNodeExecutionResponse(ResponseModel):
id: str
inputs: Any = Field(default=None, validation_alias="inputs_dict")
process_data: Any = Field(default=None, validation_alias="process_data_dict")
outputs: Any = Field(default=None, validation_alias="outputs_dict")
```
Document response models with `@ns.response(...)`:
```python
@console_ns.response(
200,
"Node run started successfully",
console_ns.models[WorkflowRunNodeExecutionResponse.__name__],
)
def post(...):
...
```
Serialize explicitly:
```python
return WorkflowRunNodeExecutionResponse.model_validate(
workflow_node_execution,
from_attributes=True,
).model_dump(mode="json")
```
If the service can return `None`, translate that into the expected HTTP error before validation:
```python
workflow_run = service.get_workflow_run(...)
if workflow_run is None:
raise NotFound("Workflow run not found")
return WorkflowRunDetailResponse.model_validate(workflow_run, from_attributes=True).model_dump(mode="json")
```
## Legacy Flask-RESTX Patterns
Avoid adding these patterns to new or migrated endpoints:
- `ns.model(...)` for new request/response DTOs.
- Module-level exported RESTX model objects such as `workflow_run_detail_model`.
- `fields.Nested({...})` with raw inline dict field maps.
- `@marshal_with(...)` for response serialization.
- `@ns.expect(...)` for GET query params.
Existing legacy field dictionaries may remain where an endpoint has not yet been migrated. Keep that compatibility local
to the legacy area and avoid importing RESTX model objects from controllers.
## Verifying Swagger
For schema and documentation changes, run focused tests and generate Swagger JSON:
```bash
uv run --project . pytest tests/unit_tests/controllers/common/test_schema.py
uv run --project . pytest tests/unit_tests/commands/test_generate_swagger_specs.py tests/unit_tests/controllers/test_swagger.py
uv run --project . dev/generate_swagger_specs.py --output-dir /tmp/dify-openapi-check
```
Inspect affected endpoints with `jq`. Check that:
- GET parameters are `in: query`.
- Request bodies appear only where the endpoint has a body.
- Responses reference the expected `*Response` schema.
- Response schemas use public serialized names, not internal validation aliases like `inputs_dict`.

30
api/controllers/common/schema.py
View File

@ -8,7 +8,7 @@ These helpers keep that translation centralized so models registered through
from collections.abc import Mapping
from enum import StrEnum
from typing import Any, NotRequired, TypedDict
from typing import Any, Literal, NotRequired, TypedDict
from flask_restx import Namespace
from pydantic import BaseModel, TypeAdapter
@ -54,16 +54,23 @@ def _register_json_schema(namespace: Namespace, name: str, schema: dict) -> None
_register_json_schema(namespace, nested_name, nested_schema)
def register_schema_model(namespace: Namespace, model: type[BaseModel]) -> None:
"""Register a BaseModel and its nested schema definitions for Swagger documentation."""
JsonSchemaMode = Literal["validation", "serialization"]
def _register_schema_model(namespace: Namespace, model: type[BaseModel], *, mode: JsonSchemaMode) -> None:
_register_json_schema(
namespace,
model.__name__,
model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0, mode=mode),
)
def register_schema_model(namespace: Namespace, model: type[BaseModel]) -> None:
"""Register a BaseModel and its nested schema definitions for Swagger documentation."""
_register_schema_model(namespace, model, mode="validation")
def register_schema_models(namespace: Namespace, *models: type[BaseModel]) -> None:
"""Register multiple BaseModels with a namespace."""
@ -71,6 +78,19 @@ def register_schema_models(namespace: Namespace, *models: type[BaseModel]) -> No
register_schema_model(namespace, model)
def register_response_schema_model(namespace: Namespace, model: type[BaseModel]) -> None:
"""Register a BaseModel using its serialized response shape."""
_register_schema_model(namespace, model, mode="serialization")
def register_response_schema_models(namespace: Namespace, *models: type[BaseModel]) -> None:
"""Register multiple response BaseModels using their serialized response shape."""
for model in models:
register_response_schema_model(namespace, model)
def get_or_create_model(model_name: str, field_def):
# Import lazily to avoid circular imports between console controllers and schema helpers.
from controllers.console import console_ns
@ -190,6 +210,8 @@ __all__ = [
"get_or_create_model",
"query_params_from_model",
"register_enum_models",
"register_response_schema_model",
"register_response_schema_models",
"register_schema_model",
"register_schema_models",
]

71
api/controllers/console/app/workflow.py
View File

@ -11,9 +11,9 @@ from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotF
import services
from controllers.common.controller_schemas import DefaultBlockConfigQuery, WorkflowListQuery, WorkflowUpdatePayload
from controllers.common.schema import register_response_schema_model, register_schema_models
from controllers.console import console_ns
from controllers.console.app.error import ConversationCompletedError, DraftWorkflowNotExist, DraftWorkflowNotSync
from controllers.console.app.workflow_run import workflow_run_node_execution_model
from controllers.console.app.wraps import get_app_model
from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
@ -37,6 +37,7 @@ from factories import file_factory, variable_factory
from fields.member_fields import simple_account_fields
from fields.online_user_fields import online_user_list_fields
from fields.workflow_fields import workflow_fields, workflow_pagination_fields
from fields.workflow_run_fields import WorkflowRunNodeExecutionResponse
from graphon.enums import NodeType
from graphon.file import File
from graphon.file import helpers as file_helpers
@ -56,6 +57,7 @@ from services.errors.llm import InvokeRateLimitError
from services.workflow_service import DraftWorkflowDeletionError, WorkflowInUseError, WorkflowService
logger = logging.getLogger(__name__)
_file_access_controller = DatabaseFileAccessController()
LISTENING_RETRY_IN = 2000
DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
@ -176,25 +178,25 @@ class DraftWorkflowTriggerRunAllPayload(BaseModel):
node_ids: list[str]
def reg(cls: type[BaseModel]):
console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
reg(SyncDraftWorkflowPayload)
reg(AdvancedChatWorkflowRunPayload)
reg(IterationNodeRunPayload)
reg(LoopNodeRunPayload)
reg(DraftWorkflowRunPayload)
reg(DraftWorkflowNodeRunPayload)
reg(PublishWorkflowPayload)
reg(DefaultBlockConfigQuery)
reg(ConvertToWorkflowPayload)
reg(WorkflowListQuery)
reg(WorkflowUpdatePayload)
reg(WorkflowFeaturesPayload)
reg(WorkflowOnlineUsersPayload)
reg(DraftWorkflowTriggerRunPayload)
reg(DraftWorkflowTriggerRunAllPayload)
register_schema_models(
console_ns,
SyncDraftWorkflowPayload,
AdvancedChatWorkflowRunPayload,
IterationNodeRunPayload,
LoopNodeRunPayload,
DraftWorkflowRunPayload,
DraftWorkflowNodeRunPayload,
PublishWorkflowPayload,
DefaultBlockConfigQuery,
ConvertToWorkflowPayload,
WorkflowListQuery,
WorkflowUpdatePayload,
WorkflowFeaturesPayload,
WorkflowOnlineUsersPayload,
DraftWorkflowTriggerRunPayload,
DraftWorkflowTriggerRunAllPayload,
)
register_response_schema_model(console_ns, WorkflowRunNodeExecutionResponse)
# TODO(QuantumGhost): Refactor existing node run API to handle file parameter parsing
@ -540,9 +542,12 @@ class HumanInputDeliveryTestPayload(BaseModel):
)
reg(HumanInputFormPreviewPayload)
reg(HumanInputFormSubmitPayload)
reg(HumanInputDeliveryTestPayload)
register_schema_models(
console_ns,
HumanInputFormPreviewPayload,
HumanInputFormSubmitPayload,
HumanInputDeliveryTestPayload,
)
@console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflows/draft/human-input/nodes/<string:node_id>/form/preview")
@ -760,14 +765,17 @@ class DraftWorkflowNodeRunApi(Resource):
@console_ns.doc(description="Run draft workflow node")
@console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
@console_ns.expect(console_ns.models[DraftWorkflowNodeRunPayload.__name__])
@console_ns.response(200, "Node run started successfully", workflow_run_node_execution_model)
@console_ns.response(
200,
"Node run started successfully",
console_ns.models[WorkflowRunNodeExecutionResponse.__name__],
)
@console_ns.response(403, "Permission denied")
@console_ns.response(404, "Node not found")
@setup_required
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
@marshal_with(workflow_run_node_execution_model)
@edit_permission_required
def post(self, app_model: App, node_id: str):
"""
@ -799,7 +807,9 @@ class DraftWorkflowNodeRunApi(Resource):
files=files,
)
return workflow_node_execution
return WorkflowRunNodeExecutionResponse.model_validate(
workflow_node_execution, from_attributes=True
).model_dump(mode="json")
@console_ns.route("/apps/<uuid:app_id>/workflows/publish")
@ -1143,14 +1153,17 @@ class DraftWorkflowNodeLastRunApi(Resource):
@console_ns.doc("get_draft_workflow_node_last_run")
@console_ns.doc(description="Get last run result for draft workflow node")
@console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
@console_ns.response(200, "Node last run retrieved successfully", workflow_run_node_execution_model)
@console_ns.response(
200,
"Node last run retrieved successfully",
console_ns.models[WorkflowRunNodeExecutionResponse.__name__],
)
@console_ns.response(404, "Node last run not found")
@console_ns.response(403, "Permission denied")
@setup_required
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
@marshal_with(workflow_run_node_execution_model)
def get(self, app_model: App, node_id: str):
srv = WorkflowService()
workflow = srv.get_draft_workflow(app_model)
@ -1163,7 +1176,7 @@ class DraftWorkflowNodeLastRunApi(Resource):
)
if node_exec is None:
raise NotFound("last run not found")
return node_exec
return WorkflowRunNodeExecutionResponse.model_validate(node_exec, from_attributes=True).model_dump(mode="json")
@console_ns.route("/apps/<uuid:app_id>/workflows/draft/trigger/run")

314
api/controllers/console/app/workflow_run.py
View File

@ -1,30 +1,28 @@
from datetime import UTC, datetime, timedelta
from typing import Literal, TypedDict, cast
from typing import Literal, cast
from flask import request
from flask_restx import Resource, fields, marshal_with
from flask_restx import Resource
from pydantic import BaseModel, Field, field_validator
from sqlalchemy import select
from sqlalchemy.orm import sessionmaker
from configs import dify_config
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 get_app_model
from controllers.console.wraps import account_initialization_required, setup_required
from controllers.web.error import NotFoundError
from core.workflow.human_input_forms import load_form_tokens_by_form_id as _load_form_tokens_by_form_id
from extensions.ext_database import db
from fields.end_user_fields import simple_end_user_fields
from fields.member_fields import simple_account_fields
from fields.base import ResponseModel
from fields.workflow_run_fields import (
advanced_chat_workflow_run_for_list_fields,
advanced_chat_workflow_run_pagination_fields,
workflow_run_count_fields,
workflow_run_detail_fields,
workflow_run_for_list_fields,
workflow_run_node_execution_fields,
workflow_run_node_execution_list_fields,
workflow_run_pagination_fields,
AdvancedChatWorkflowRunPaginationResponse,
WorkflowRunCountResponse,
WorkflowRunDetailResponse,
WorkflowRunNodeExecutionListResponse,
WorkflowRunNodeExecutionResponse,
WorkflowRunPaginationResponse,
)
from graphon.entities.pause_reason import HumanInputRequired
from graphon.enums import WorkflowExecutionStatus
@ -52,82 +50,6 @@ def _build_backstage_input_url(form_token: str | None) -> str | None:
WORKFLOW_RUN_STATUS_CHOICES = ["running", "succeeded", "failed", "stopped", "partial-succeeded"]
EXPORT_SIGNED_URL_EXPIRE_SECONDS = 3600
# Register models for flask_restx to avoid dict type issues in Swagger
# Register in dependency order: base models first, then dependent models
# Base models
simple_account_model = console_ns.model("SimpleAccount", simple_account_fields)
simple_end_user_model = console_ns.model("SimpleEndUser", simple_end_user_fields)
# Models that depend on simple_account_fields
workflow_run_for_list_fields_copy = workflow_run_for_list_fields.copy()
workflow_run_for_list_fields_copy["created_by_account"] = fields.Nested(
simple_account_model, attribute="created_by_account", allow_null=True
)
workflow_run_for_list_model = console_ns.model("WorkflowRunForList", workflow_run_for_list_fields_copy)
advanced_chat_workflow_run_for_list_fields_copy = advanced_chat_workflow_run_for_list_fields.copy()
advanced_chat_workflow_run_for_list_fields_copy["created_by_account"] = fields.Nested(
simple_account_model, attribute="created_by_account", allow_null=True
)
advanced_chat_workflow_run_for_list_model = console_ns.model(
"AdvancedChatWorkflowRunForList", advanced_chat_workflow_run_for_list_fields_copy
)
workflow_run_detail_fields_copy = workflow_run_detail_fields.copy()
workflow_run_detail_fields_copy["created_by_account"] = fields.Nested(
simple_account_model, attribute="created_by_account", allow_null=True
)
workflow_run_detail_fields_copy["created_by_end_user"] = fields.Nested(
simple_end_user_model, attribute="created_by_end_user", allow_null=True
)
workflow_run_detail_model = console_ns.model("WorkflowRunDetail", workflow_run_detail_fields_copy)
workflow_run_node_execution_fields_copy = workflow_run_node_execution_fields.copy()
workflow_run_node_execution_fields_copy["created_by_account"] = fields.Nested(
simple_account_model, attribute="created_by_account", allow_null=True
)
workflow_run_node_execution_fields_copy["created_by_end_user"] = fields.Nested(
simple_end_user_model, attribute="created_by_end_user", allow_null=True
)
workflow_run_node_execution_model = console_ns.model(
"WorkflowRunNodeExecution", workflow_run_node_execution_fields_copy
)
# Simple models without nested dependencies
workflow_run_count_model = console_ns.model("WorkflowRunCount", workflow_run_count_fields)
# Pagination models that depend on list models
advanced_chat_workflow_run_pagination_fields_copy = advanced_chat_workflow_run_pagination_fields.copy()
advanced_chat_workflow_run_pagination_fields_copy["data"] = fields.List(
fields.Nested(advanced_chat_workflow_run_for_list_model), attribute="data"
)
advanced_chat_workflow_run_pagination_model = console_ns.model(
"AdvancedChatWorkflowRunPagination", advanced_chat_workflow_run_pagination_fields_copy
)
workflow_run_pagination_fields_copy = workflow_run_pagination_fields.copy()
workflow_run_pagination_fields_copy["data"] = fields.List(fields.Nested(workflow_run_for_list_model), attribute="data")
workflow_run_pagination_model = console_ns.model("WorkflowRunPagination", workflow_run_pagination_fields_copy)
workflow_run_node_execution_list_fields_copy = workflow_run_node_execution_list_fields.copy()
workflow_run_node_execution_list_fields_copy["data"] = fields.List(fields.Nested(workflow_run_node_execution_model))
workflow_run_node_execution_list_model = console_ns.model(
"WorkflowRunNodeExecutionList", workflow_run_node_execution_list_fields_copy
)
workflow_run_export_fields = console_ns.model(
"WorkflowRunExport",
{
"status": fields.String(description="Export status: success/failed"),
"presigned_url": fields.String(description="Pre-signed URL for download", required=False),
"presigned_url_expires_at": fields.String(description="Pre-signed URL expiration time", required=False),
},
)
DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
class WorkflowRunListQuery(BaseModel):
last_id: str | None = Field(default=None, description="Last run ID for pagination")
@ -136,7 +58,7 @@ class WorkflowRunListQuery(BaseModel):
default=None, description="Workflow run status filter"
)
triggered_from: Literal["debugging", "app-run"] | None = Field(
default=None, description="Filter by trigger source: debugging or app-run"
default=None, description="Filter by trigger source: debugging or app-run. Default: debugging"
)
@field_validator("last_id")
@ -151,9 +73,15 @@ class WorkflowRunCountQuery(BaseModel):
status: Literal["running", "succeeded", "failed", "stopped", "partial-succeeded"] | None = Field(
default=None, description="Workflow run status filter"
)
time_range: str | None = Field(default=None, description="Time range filter (e.g., 7d, 4h, 30m, 30s)")
time_range: str | None = Field(
default=None,
description=(
"Filter by time range (optional): e.g., 7d (7 days), 4h (4 hours), "
"30m (30 minutes), 30s (30 seconds). Filters by created_at field."
),
)
triggered_from: Literal["debugging", "app-run"] | None = Field(
default=None, description="Filter by trigger source: debugging or app-run"
default=None, description="Filter by trigger source: debugging or app-run. Default: debugging"
)
@field_validator("time_range")
@ -164,51 +92,64 @@ class WorkflowRunCountQuery(BaseModel):
return time_duration(value)
console_ns.schema_model(
WorkflowRunListQuery.__name__, WorkflowRunListQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
)
console_ns.schema_model(
WorkflowRunCountQuery.__name__,
WorkflowRunCountQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
)
class WorkflowRunExportResponse(ResponseModel):
status: str = Field(description="Export status: success/failed")
presigned_url: str | None = Field(default=None, description="Pre-signed URL for download")
presigned_url_expires_at: str | None = Field(default=None, description="Pre-signed URL expiration time")
class HumanInputPauseTypeResponse(TypedDict):
class HumanInputPauseTypeResponse(ResponseModel):
type: Literal["human_input"]
form_id: str
backstage_input_url: str | None
backstage_input_url: str | None = None
class PausedNodeResponse(TypedDict):
class PausedNodeResponse(ResponseModel):
node_id: str
node_title: str
pause_type: HumanInputPauseTypeResponse
class WorkflowPauseDetailsResponse(TypedDict):
paused_at: str | None
class WorkflowPauseDetailsResponse(ResponseModel):
paused_at: str | None = None
paused_nodes: list[PausedNodeResponse]
register_schema_models(
console_ns,
WorkflowRunListQuery,
WorkflowRunCountQuery,
)
register_response_schema_models(
console_ns,
AdvancedChatWorkflowRunPaginationResponse,
WorkflowRunPaginationResponse,
WorkflowRunCountResponse,
WorkflowRunDetailResponse,
WorkflowRunNodeExecutionResponse,
WorkflowRunNodeExecutionListResponse,
WorkflowRunExportResponse,
HumanInputPauseTypeResponse,
PausedNodeResponse,
WorkflowPauseDetailsResponse,
)
@console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflow-runs")
class AdvancedChatAppWorkflowRunListApi(Resource):
@console_ns.doc("get_advanced_chat_workflow_runs")
@console_ns.doc(description="Get advanced chat workflow run list")
@console_ns.doc(params={"app_id": "Application ID"})
@console_ns.doc(params={"last_id": "Last run ID for pagination", "limit": "Number of items per page (1-100)"})
@console_ns.doc(
params={"status": "Filter by status (optional): running, succeeded, failed, stopped, partial-succeeded"}
@console_ns.doc(params=query_params_from_model(WorkflowRunListQuery))
@console_ns.response(
200,
"Workflow runs retrieved successfully",
console_ns.models[AdvancedChatWorkflowRunPaginationResponse.__name__],
)
@console_ns.doc(
params={"triggered_from": "Filter by trigger source (optional): debugging or app-run. Default: debugging"}
)
@console_ns.expect(console_ns.models[WorkflowRunListQuery.__name__])
@console_ns.response(200, "Workflow runs retrieved successfully", advanced_chat_workflow_run_pagination_model)
@setup_required
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.ADVANCED_CHAT])
@marshal_with(advanced_chat_workflow_run_pagination_model)
def get(self, app_model: App):
"""
Get advanced chat app workflow run list
@ -232,7 +173,9 @@ class AdvancedChatAppWorkflowRunListApi(Resource):
app_model=app_model, args=args, triggered_from=triggered_from
)
return result
return AdvancedChatWorkflowRunPaginationResponse.model_validate(result, from_attributes=True).model_dump(
mode="json"
)
@console_ns.route("/apps/<uuid:app_id>/workflow-runs/<uuid:run_id>/export")
@ -240,7 +183,7 @@ class WorkflowRunExportApi(Resource):
@console_ns.doc("get_workflow_run_export_url")
@console_ns.doc(description="Generate a download URL for an archived workflow run.")
@console_ns.doc(params={"app_id": "Application ID", "run_id": "Workflow run ID"})
@console_ns.response(200, "Export URL generated", workflow_run_export_fields)
@console_ns.response(200, "Export URL generated", console_ns.models[WorkflowRunExportResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@ -278,11 +221,14 @@ class WorkflowRunExportApi(Resource):
expires_in=EXPORT_SIGNED_URL_EXPIRE_SECONDS,
)
expires_at = datetime.now(UTC) + timedelta(seconds=EXPORT_SIGNED_URL_EXPIRE_SECONDS)
return {
"status": "success",
"presigned_url": presigned_url,
"presigned_url_expires_at": expires_at.isoformat(),
}, 200
response = WorkflowRunExportResponse.model_validate(
{
"status": "success",
"presigned_url": presigned_url,
"presigned_url_expires_at": expires_at.isoformat(),
}
)
return response.model_dump(mode="json"), 200
@console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflow-runs/count")
@ -290,27 +236,16 @@ class AdvancedChatAppWorkflowRunCountApi(Resource):
@console_ns.doc("get_advanced_chat_workflow_runs_count")
@console_ns.doc(description="Get advanced chat workflow runs count statistics")
@console_ns.doc(params={"app_id": "Application ID"})
@console_ns.doc(
params={"status": "Filter by status (optional): running, succeeded, failed, stopped, partial-succeeded"}
@console_ns.doc(params=query_params_from_model(WorkflowRunCountQuery))
@console_ns.response(
200,
"Workflow runs count retrieved successfully",
console_ns.models[WorkflowRunCountResponse.__name__],
)
@console_ns.doc(
params={
"time_range": (
"Filter by time range (optional): e.g., 7d (7 days), 4h (4 hours), "
"30m (30 minutes), 30s (30 seconds). Filters by created_at field."
)
}
)
@console_ns.doc(
params={"triggered_from": "Filter by trigger source (optional): debugging or app-run. Default: debugging"}
)
@console_ns.response(200, "Workflow runs count retrieved successfully", workflow_run_count_model)
@console_ns.expect(console_ns.models[WorkflowRunCountQuery.__name__])
@setup_required
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.ADVANCED_CHAT])
@marshal_with(workflow_run_count_model)
def get(self, app_model: App):
"""
Get advanced chat workflow runs count statistics
@ -333,7 +268,7 @@ class AdvancedChatAppWorkflowRunCountApi(Resource):
triggered_from=triggered_from,
)
return result
return WorkflowRunCountResponse.model_validate(result).model_dump(mode="json")
@console_ns.route("/apps/<uuid:app_id>/workflow-runs")
@ -341,20 +276,16 @@ class WorkflowRunListApi(Resource):
@console_ns.doc("get_workflow_runs")
@console_ns.doc(description="Get workflow run list")
@console_ns.doc(params={"app_id": "Application ID"})
@console_ns.doc(params={"last_id": "Last run ID for pagination", "limit": "Number of items per page (1-100)"})
@console_ns.doc(
params={"status": "Filter by status (optional): running, succeeded, failed, stopped, partial-succeeded"}
@console_ns.doc(params=query_params_from_model(WorkflowRunListQuery))
@console_ns.response(
200,
"Workflow runs retrieved successfully",
console_ns.models[WorkflowRunPaginationResponse.__name__],
)
@console_ns.doc(
params={"triggered_from": "Filter by trigger source (optional): debugging or app-run. Default: debugging"}
)
@console_ns.response(200, "Workflow runs retrieved successfully", workflow_run_pagination_model)
@console_ns.expect(console_ns.models[WorkflowRunListQuery.__name__])
@setup_required
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
@marshal_with(workflow_run_pagination_model)
def get(self, app_model: App):
"""
Get workflow run list
@ -378,7 +309,7 @@ class WorkflowRunListApi(Resource):
app_model=app_model, args=args, triggered_from=triggered_from
)
return result
return WorkflowRunPaginationResponse.model_validate(result, from_attributes=True).model_dump(mode="json")
@console_ns.route("/apps/<uuid:app_id>/workflow-runs/count")
@ -386,27 +317,16 @@ class WorkflowRunCountApi(Resource):
@console_ns.doc("get_workflow_runs_count")
@console_ns.doc(description="Get workflow runs count statistics")
@console_ns.doc(params={"app_id": "Application ID"})
@console_ns.doc(
params={"status": "Filter by status (optional): running, succeeded, failed, stopped, partial-succeeded"}
@console_ns.doc(params=query_params_from_model(WorkflowRunCountQuery))
@console_ns.response(
200,
"Workflow runs count retrieved successfully",
console_ns.models[WorkflowRunCountResponse.__name__],
)
@console_ns.doc(
params={
"time_range": (
"Filter by time range (optional): e.g., 7d (7 days), 4h (4 hours), "
"30m (30 minutes), 30s (30 seconds). Filters by created_at field."
)
}
)
@console_ns.doc(
params={"triggered_from": "Filter by trigger source (optional): debugging or app-run. Default: debugging"}
)
@console_ns.response(200, "Workflow runs count retrieved successfully", workflow_run_count_model)
@console_ns.expect(console_ns.models[WorkflowRunCountQuery.__name__])
@setup_required
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
@marshal_with(workflow_run_count_model)
def get(self, app_model: App):
"""
Get workflow runs count statistics
@ -429,7 +349,7 @@ class WorkflowRunCountApi(Resource):
triggered_from=triggered_from,
)
return result
return WorkflowRunCountResponse.model_validate(result).model_dump(mode="json")
@console_ns.route("/apps/<uuid:app_id>/workflow-runs/<uuid:run_id>")
@ -437,13 +357,16 @@ class WorkflowRunDetailApi(Resource):
@console_ns.doc("get_workflow_run_detail")
@console_ns.doc(description="Get workflow run detail")
@console_ns.doc(params={"app_id": "Application ID", "run_id": "Workflow run ID"})
@console_ns.response(200, "Workflow run detail retrieved successfully", workflow_run_detail_model)
@console_ns.response(
200,
"Workflow run detail retrieved successfully",
console_ns.models[WorkflowRunDetailResponse.__name__],
)
@console_ns.response(404, "Workflow run not found")
@setup_required
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
@marshal_with(workflow_run_detail_model)
def get(self, app_model: App, run_id):
"""
Get workflow run detail
@ -452,8 +375,10 @@ class WorkflowRunDetailApi(Resource):
workflow_run_service = WorkflowRunService()
workflow_run = workflow_run_service.get_workflow_run(app_model=app_model, run_id=run_id)
if workflow_run is None:
raise NotFoundError("Workflow run not found")
return workflow_run
return WorkflowRunDetailResponse.model_validate(workflow_run, from_attributes=True).model_dump(mode="json")
@console_ns.route("/apps/<uuid:app_id>/workflow-runs/<uuid:run_id>/node-executions")
@ -461,13 +386,16 @@ class WorkflowRunNodeExecutionListApi(Resource):
@console_ns.doc("get_workflow_run_node_executions")
@console_ns.doc(description="Get workflow run node execution list")
@console_ns.doc(params={"app_id": "Application ID", "run_id": "Workflow run ID"})
@console_ns.response(200, "Node executions retrieved successfully", workflow_run_node_execution_list_model)
@console_ns.response(
200,
"Node executions retrieved successfully",
console_ns.models[WorkflowRunNodeExecutionListResponse.__name__],
)
@console_ns.response(404, "Workflow run not found")
@setup_required
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
@marshal_with(workflow_run_node_execution_list_model)
def get(self, app_model: App, run_id):
"""
Get workflow run node execution list
@ -482,13 +410,24 @@ class WorkflowRunNodeExecutionListApi(Resource):
user=user,
)
return {"data": node_executions}
return WorkflowRunNodeExecutionListResponse.model_validate(
{"data": node_executions}, from_attributes=True
).model_dump(mode="json")
@console_ns.route("/workflow/<string:workflow_run_id>/pause-details")
class ConsoleWorkflowPauseDetailsApi(Resource):
"""Console API for getting workflow pause details."""
@console_ns.doc("get_workflow_pause_details")
@console_ns.doc(description="Get workflow pause details")
@console_ns.doc(params={"workflow_run_id": "Workflow run ID"})
@console_ns.response(
200,
"Workflow pause details retrieved successfully",
console_ns.models[WorkflowPauseDetailsResponse.__name__],
)
@console_ns.response(404, "Workflow run not found")
@setup_required
@login_required
@account_initialization_required
@ -515,11 +454,8 @@ class ConsoleWorkflowPauseDetailsApi(Resource):
# Check if workflow is suspended
is_paused = workflow_run.status == WorkflowExecutionStatus.PAUSED
if not is_paused:
empty_response: WorkflowPauseDetailsResponse = {
"paused_at": None,
"paused_nodes": [],
}
return empty_response, 200
empty_response = WorkflowPauseDetailsResponse(paused_at=None, paused_nodes=[])
return empty_response.model_dump(mode="json"), 200
pause_entity = workflow_run_repo.get_workflow_pause(workflow_run_id)
pause_reasons = pause_entity.get_pause_reasons() if pause_entity else []
@ -530,27 +466,25 @@ class ConsoleWorkflowPauseDetailsApi(Resource):
# Build response
paused_at = pause_entity.paused_at if pause_entity else None
paused_nodes: list[PausedNodeResponse] = []
response: WorkflowPauseDetailsResponse = {
"paused_at": paused_at.isoformat() + "Z" if paused_at else None,
"paused_nodes": paused_nodes,
}
for reason in pause_reasons:
if isinstance(reason, HumanInputRequired):
paused_nodes.append(
{
"node_id": reason.node_id,
"node_title": reason.node_title,
"pause_type": {
"type": "human_input",
"form_id": reason.form_id,
"backstage_input_url": _build_backstage_input_url(
form_tokens_by_form_id.get(reason.form_id)
),
},
}
PausedNodeResponse(
node_id=reason.node_id,
node_title=reason.node_title,
pause_type=HumanInputPauseTypeResponse(
type="human_input",
form_id=reason.form_id,
backstage_input_url=_build_backstage_input_url(form_tokens_by_form_id.get(reason.form_id)),
),
)
)
else:
raise AssertionError("unimplemented.")
return response, 200
response = WorkflowPauseDetailsResponse(
paused_at=paused_at.isoformat() + "Z" if paused_at else None,
paused_nodes=paused_nodes,
)
return response.model_dump(mode="json"), 200

77
api/controllers/console/datasets/rag_pipeline/rag_pipeline_workflow.py
View File

@ -10,7 +10,7 @@ from werkzeug.exceptions import BadRequest, Forbidden, InternalServerError, NotF
import services
from controllers.common.controller_schemas import DefaultBlockConfigQuery, WorkflowListQuery, WorkflowUpdatePayload
from controllers.common.schema import register_schema_models
from controllers.common.schema import register_response_schema_models, register_schema_models
from controllers.console import console_ns
from controllers.console.app.error import (
ConversationCompletedError,
@ -22,12 +22,6 @@ from controllers.console.app.workflow import (
workflow_model,
workflow_pagination_model,
)
from controllers.console.app.workflow_run import (
workflow_run_detail_model,
workflow_run_node_execution_list_model,
workflow_run_node_execution_model,
workflow_run_pagination_model,
)
from controllers.console.datasets.wraps import get_rag_pipeline
from controllers.console.wraps import (
account_initialization_required,
@ -40,6 +34,12 @@ from core.app.apps.pipeline.pipeline_generator import PipelineGenerator
from core.app.entities.app_invoke_entities import InvokeFrom
from extensions.ext_database import db
from factories import variable_factory
from fields.workflow_run_fields import (
WorkflowRunDetailResponse,
WorkflowRunNodeExecutionListResponse,
WorkflowRunNodeExecutionResponse,
WorkflowRunPaginationResponse,
)
from graphon.model_runtime.utils.encoders import jsonable_encoder
from libs import helper
from libs.helper import TimestampField, UUIDStrOrEmpty
@ -131,6 +131,13 @@ register_schema_models(
DatasourceVariablesPayload,
RagPipelineRecommendedPluginQuery,
)
register_response_schema_models(
console_ns,
WorkflowRunDetailResponse,
WorkflowRunNodeExecutionListResponse,
WorkflowRunNodeExecutionResponse,
WorkflowRunPaginationResponse,
)
@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft")
@ -415,12 +422,16 @@ class RagPipelineDraftDatasourceNodeRunApi(Resource):
@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/nodes/<string:node_id>/run")
class RagPipelineDraftNodeRunApi(Resource):
@console_ns.expect(console_ns.models[NodeRunRequiredPayload.__name__])
@console_ns.response(
200,
"Node run started successfully",
console_ns.models[WorkflowRunNodeExecutionResponse.__name__],
)
@setup_required
@login_required
@edit_permission_required
@account_initialization_required
@get_rag_pipeline
@marshal_with(workflow_run_node_execution_model)
def post(self, pipeline: Pipeline, node_id: str):
"""
Run draft workflow node
@ -439,7 +450,9 @@ class RagPipelineDraftNodeRunApi(Resource):
if workflow_node_execution is None:
raise ValueError("Workflow node execution not found")
return workflow_node_execution
return WorkflowRunNodeExecutionResponse.model_validate(
workflow_node_execution, from_attributes=True
).model_dump(mode="json")
@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflow-runs/tasks/<string:task_id>/stop")
@ -778,11 +791,15 @@ class DraftRagPipelineSecondStepApi(Resource):
@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflow-runs")
class RagPipelineWorkflowRunListApi(Resource):
@console_ns.response(
200,
"Workflow runs retrieved successfully",
console_ns.models[WorkflowRunPaginationResponse.__name__],
)
@setup_required
@login_required
@account_initialization_required
@get_rag_pipeline
@marshal_with(workflow_run_pagination_model)
def get(self, pipeline: Pipeline):
"""
Get workflow run list
@ -801,16 +818,20 @@ class RagPipelineWorkflowRunListApi(Resource):
rag_pipeline_service = RagPipelineService()
result = rag_pipeline_service.get_rag_pipeline_paginate_workflow_runs(pipeline=pipeline, args=args)
return result
return WorkflowRunPaginationResponse.model_validate(result, from_attributes=True).model_dump(mode="json")
@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflow-runs/<uuid:run_id>")
class RagPipelineWorkflowRunDetailApi(Resource):
@console_ns.response(
200,
"Workflow run detail retrieved successfully",
console_ns.models[WorkflowRunDetailResponse.__name__],
)
@setup_required
@login_required
@account_initialization_required
@get_rag_pipeline
@marshal_with(workflow_run_detail_model)
def get(self, pipeline: Pipeline, run_id):
"""
Get workflow run detail
@ -819,17 +840,23 @@ class RagPipelineWorkflowRunDetailApi(Resource):
rag_pipeline_service = RagPipelineService()
workflow_run = rag_pipeline_service.get_rag_pipeline_workflow_run(pipeline=pipeline, run_id=run_id)
if workflow_run is None:
raise NotFound("Workflow run not found")
return workflow_run
return WorkflowRunDetailResponse.model_validate(workflow_run, from_attributes=True).model_dump(mode="json")
@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflow-runs/<uuid:run_id>/node-executions")
class RagPipelineWorkflowRunNodeExecutionListApi(Resource):
@console_ns.response(
200,
"Node executions retrieved successfully",
console_ns.models[WorkflowRunNodeExecutionListResponse.__name__],
)
@setup_required
@login_required
@account_initialization_required
@get_rag_pipeline
@marshal_with(workflow_run_node_execution_list_model)
def get(self, pipeline: Pipeline, run_id: str):
"""
Get workflow run node execution list
@ -844,7 +871,9 @@ class RagPipelineWorkflowRunNodeExecutionListApi(Resource):
user=user,
)
return {"data": node_executions}
return WorkflowRunNodeExecutionListResponse.model_validate(
{"data": node_executions}, from_attributes=True
).model_dump(mode="json")
@console_ns.route("/rag/pipelines/datasource-plugins")
@ -859,11 +888,15 @@ class DatasourceListApi(Resource):
@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/nodes/<string:node_id>/last-run")
class RagPipelineWorkflowLastRunApi(Resource):
@console_ns.response(
200,
"Node last run retrieved successfully",
console_ns.models[WorkflowRunNodeExecutionResponse.__name__],
)
@setup_required
@login_required
@account_initialization_required
@get_rag_pipeline
@marshal_with(workflow_run_node_execution_model)
def get(self, pipeline: Pipeline, node_id: str):
rag_pipeline_service = RagPipelineService()
workflow = rag_pipeline_service.get_draft_workflow(pipeline=pipeline)
@ -876,7 +909,7 @@ class RagPipelineWorkflowLastRunApi(Resource):
)
if node_exec is None:
raise NotFound("last run not found")
return node_exec
return WorkflowRunNodeExecutionResponse.model_validate(node_exec, from_attributes=True).model_dump(mode="json")
@console_ns.route("/rag/pipelines/transform/datasets/<uuid:dataset_id>")
@ -899,12 +932,16 @@ class RagPipelineTransformApi(Resource):
@console_ns.route("/rag/pipelines/<uuid:pipeline_id>/workflows/draft/datasource/variables-inspect")
class RagPipelineDatasourceVariableApi(Resource):
@console_ns.expect(console_ns.models[DatasourceVariablesPayload.__name__])
@console_ns.response(
200,
"Datasource variables set successfully",
console_ns.models[WorkflowRunNodeExecutionResponse.__name__],
)
@setup_required
@login_required
@account_initialization_required
@get_rag_pipeline
@edit_permission_required
@marshal_with(workflow_run_node_execution_model)
def post(self, pipeline: Pipeline):
"""
Set datasource variables
@ -918,7 +955,9 @@ class RagPipelineDatasourceVariableApi(Resource):
args=args,
current_user=current_user,
)
return workflow_node_execution
return WorkflowRunNodeExecutionResponse.model_validate(
workflow_node_execution, from_attributes=True
).model_dump(mode="json")
@console_ns.route("/rag/pipelines/recommended-plugins")

131
api/fields/workflow_run_fields.py
View File

@ -1,14 +1,21 @@
"""Workflow run response schemas for console APIs.
Most workflow-run endpoints should document and serialize responses with the
Pydantic models in this module. The remaining Flask-RESTX field dictionaries are
kept only for workflow app-log endpoints that still build legacy log models.
"""
from __future__ import annotations
from datetime import datetime
from typing import Any
from flask_restx import Namespace, fields
from pydantic import Field, field_validator
from pydantic import AliasChoices, Field, field_validator
from fields.base import ResponseModel
from fields.end_user_fields import SimpleEndUser, simple_end_user_fields
from fields.member_fields import SimpleAccount, simple_account_fields
from fields.end_user_fields import SimpleEndUser
from fields.member_fields import SimpleAccount
from libs.helper import TimestampField
workflow_run_for_log_fields = {
@ -43,119 +50,6 @@ def build_workflow_run_for_archived_log_model(api_or_ns: Namespace):
return api_or_ns.model("WorkflowRunForArchivedLog", workflow_run_for_archived_log_fields)
workflow_run_for_list_fields = {
"id": fields.String,
"version": fields.String,
"status": fields.String,
"elapsed_time": fields.Float,
"total_tokens": fields.Integer,
"total_steps": fields.Integer,
"created_by_account": fields.Nested(simple_account_fields, attribute="created_by_account", allow_null=True),
"created_at": TimestampField,
"finished_at": TimestampField,
"exceptions_count": fields.Integer,
"retry_index": fields.Integer,
}
advanced_chat_workflow_run_for_list_fields = {
"id": fields.String,
"conversation_id": fields.String,
"message_id": fields.String,
"version": fields.String,
"status": fields.String,
"elapsed_time": fields.Float,
"total_tokens": fields.Integer,
"total_steps": fields.Integer,
"created_by_account": fields.Nested(simple_account_fields, attribute="created_by_account", allow_null=True),
"created_at": TimestampField,
"finished_at": TimestampField,
"exceptions_count": fields.Integer,
"retry_index": fields.Integer,
}
advanced_chat_workflow_run_pagination_fields = {
"limit": fields.Integer(attribute="limit"),
"has_more": fields.Boolean(attribute="has_more"),
"data": fields.List(fields.Nested(advanced_chat_workflow_run_for_list_fields), attribute="data"),
}
workflow_run_pagination_fields = {
"limit": fields.Integer(attribute="limit"),
"has_more": fields.Boolean(attribute="has_more"),
"data": fields.List(fields.Nested(workflow_run_for_list_fields), attribute="data"),
}
workflow_run_count_fields = {
"total": fields.Integer,
"running": fields.Integer,
"succeeded": fields.Integer,
"failed": fields.Integer,
"stopped": fields.Integer,
"partial_succeeded": fields.Integer(attribute="partial-succeeded"),
}
workflow_run_detail_fields = {
"id": fields.String,
"version": fields.String,
"graph": fields.Raw(attribute="graph_dict"),
"inputs": fields.Raw(attribute="inputs_dict"),
"status": fields.String,
"outputs": fields.Raw(attribute="outputs_dict"),
"error": fields.String,
"elapsed_time": fields.Float,
"total_tokens": fields.Integer,
"total_steps": fields.Integer,
"created_by_role": fields.String,
"created_by_account": fields.Nested(simple_account_fields, attribute="created_by_account", allow_null=True),
"created_by_end_user": fields.Nested(simple_end_user_fields, attribute="created_by_end_user", allow_null=True),
"created_at": TimestampField,
"finished_at": TimestampField,
"exceptions_count": fields.Integer,
}
retry_event_field = {
"elapsed_time": fields.Float,
"status": fields.String,
"inputs": fields.Raw(attribute="inputs"),
"process_data": fields.Raw(attribute="process_data"),
"outputs": fields.Raw(attribute="outputs"),
"metadata": fields.Raw(attribute="metadata"),
"llm_usage": fields.Raw(attribute="llm_usage"),
"error": fields.String,
"retry_index": fields.Integer,
}
workflow_run_node_execution_fields = {
"id": fields.String,
"index": fields.Integer,
"predecessor_node_id": fields.String,
"node_id": fields.String,
"node_type": fields.String,
"title": fields.String,
"inputs": fields.Raw(attribute="inputs_dict"),
"process_data": fields.Raw(attribute="process_data_dict"),
"outputs": fields.Raw(attribute="outputs_dict"),
"status": fields.String,
"error": fields.String,
"elapsed_time": fields.Float,
"execution_metadata": fields.Raw(attribute="execution_metadata_dict"),
"extras": fields.Raw,
"created_at": TimestampField,
"created_by_role": fields.String,
"created_by_account": fields.Nested(simple_account_fields, attribute="created_by_account", allow_null=True),
"created_by_end_user": fields.Nested(simple_end_user_fields, attribute="created_by_end_user", allow_null=True),
"finished_at": TimestampField,
"inputs_truncated": fields.Boolean,
"outputs_truncated": fields.Boolean,
"process_data_truncated": fields.Boolean,
}
workflow_run_node_execution_list_fields = {
"data": fields.List(fields.Nested(workflow_run_node_execution_fields)),
}
def _to_timestamp(value: datetime | int | None) -> int | None:
if isinstance(value, datetime):
return int(value.timestamp())
@ -252,7 +146,10 @@ class WorkflowRunCountResponse(ResponseModel):
succeeded: int
failed: int
stopped: int
partial_succeeded: int = Field(validation_alias="partial-succeeded")
partial_succeeded: int = Field(
alias="partial_succeeded",
validation_alias=AliasChoices("partial_succeeded", "partial-succeeded"),
)
class WorkflowRunDetailResponse(ResponseModel):

291
api/openapi/markdown/console-swagger.md
View File

@ -805,18 +805,17 @@ Get advanced chat workflow run list
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| payload | body | | Yes | [WorkflowRunListQuery](#workflowrunlistquery) |
| app_id | path | Application ID | Yes | string |
| last_id | query | Last run ID for pagination | No | string |
| limit | query | Number of items per page (1-100) | No | string |
| status | query | Filter by status (optional): running, succeeded, failed, stopped, partial-succeeded | No | string |
| triggered_from | query | Filter by trigger source (optional): debugging or app-run. Default: debugging | No | string |
| limit | query | Number of items per page (1-100) | No | integer |
| status | query | Workflow run status filter | No | string |
| triggered_from | query | Filter by trigger source: debugging or app-run. Default: debugging | No | string |
##### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Workflow runs retrieved successfully | [AdvancedChatWorkflowRunPagination](#advancedchatworkflowrunpagination) |
| 200 | Workflow runs retrieved successfully | [AdvancedChatWorkflowRunPaginationResponse](#advancedchatworkflowrunpaginationresponse) |
### /apps/{app_id}/advanced-chat/workflow-runs/count
@ -833,17 +832,16 @@ Get advanced chat workflow runs count statistics
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| payload | body | | Yes | [WorkflowRunCountQuery](#workflowruncountquery) |
| app_id | path | Application ID | Yes | string |
| status | query | Filter by status (optional): running, succeeded, failed, stopped, partial-succeeded | No | string |
| status | query | Workflow run status filter | No | string |
| time_range | query | Filter by time range (optional): e.g., 7d (7 days), 4h (4 hours), 30m (30 minutes), 30s (30 seconds). Filters by created_at field. | No | string |
| triggered_from | query | Filter by trigger source (optional): debugging or app-run. Default: debugging | No | string |
| triggered_from | query | Filter by trigger source: debugging or app-run. Default: debugging | No | string |
##### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Workflow runs count retrieved successfully | [WorkflowRunCount](#workflowruncount) |
| 200 | Workflow runs count retrieved successfully | [WorkflowRunCountResponse](#workflowruncountresponse) |
### /apps/{app_id}/advanced-chat/workflows/draft/human-input/nodes/{node_id}/form/preview
@ -2361,18 +2359,17 @@ Get workflow run list
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| payload | body | | Yes | [WorkflowRunListQuery](#workflowrunlistquery) |
| app_id | path | Application ID | Yes | string |
| last_id | query | Last run ID for pagination | No | string |
| limit | query | Number of items per page (1-100) | No | string |
| status | query | Filter by status (optional): running, succeeded, failed, stopped, partial-succeeded | No | string |
| triggered_from | query | Filter by trigger source (optional): debugging or app-run. Default: debugging | No | string |
| limit | query | Number of items per page (1-100) | No | integer |
| status | query | Workflow run status filter | No | string |
| triggered_from | query | Filter by trigger source: debugging or app-run. Default: debugging | No | string |
##### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Workflow runs retrieved successfully | [WorkflowRunPagination](#workflowrunpagination) |
| 200 | Workflow runs retrieved successfully | [WorkflowRunPaginationResponse](#workflowrunpaginationresponse) |
### /apps/{app_id}/workflow-runs/count
@ -2389,17 +2386,16 @@ Get workflow runs count statistics
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| payload | body | | Yes | [WorkflowRunCountQuery](#workflowruncountquery) |
| app_id | path | Application ID | Yes | string |
| status | query | Filter by status (optional): running, succeeded, failed, stopped, partial-succeeded | No | string |
| status | query | Workflow run status filter | No | string |
| time_range | query | Filter by time range (optional): e.g., 7d (7 days), 4h (4 hours), 30m (30 minutes), 30s (30 seconds). Filters by created_at field. | No | string |
| triggered_from | query | Filter by trigger source (optional): debugging or app-run. Default: debugging | No | string |
| triggered_from | query | Filter by trigger source: debugging or app-run. Default: debugging | No | string |
##### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Workflow runs count retrieved successfully | [WorkflowRunCount](#workflowruncount) |
| 200 | Workflow runs count retrieved successfully | [WorkflowRunCountResponse](#workflowruncountresponse) |
### /apps/{app_id}/workflow-runs/tasks/{task_id}/stop
@ -2449,7 +2445,7 @@ Get workflow run detail
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Workflow run detail retrieved successfully | [WorkflowRunDetail](#workflowrundetail) |
| 200 | Workflow run detail retrieved successfully | [WorkflowRunDetailResponse](#workflowrundetailresponse) |
| 404 | Workflow run not found | |
### /apps/{app_id}/workflow-runs/{run_id}/export
@ -2470,7 +2466,7 @@ Generate a download URL for an archived workflow run.
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Export URL generated | [WorkflowRunExport](#workflowrunexport) |
| 200 | Export URL generated | [WorkflowRunExportResponse](#workflowrunexportresponse) |
### /apps/{app_id}/workflow-runs/{run_id}/node-executions
@ -2494,7 +2490,7 @@ Get workflow run node execution list
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Node executions retrieved successfully | [WorkflowRunNodeExecutionList](#workflowrunnodeexecutionlist) |
| 200 | Node executions retrieved successfully | [WorkflowRunNodeExecutionListResponse](#workflowrunnodeexecutionlistresponse) |
| 404 | Workflow run not found | |
### /apps/{app_id}/workflow/comments
@ -3180,7 +3176,7 @@ Get last run result for draft workflow node
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Node last run retrieved successfully | [WorkflowRunNodeExecution](#workflowrunnodeexecution) |
| 200 | Node last run retrieved successfully | [WorkflowRunNodeExecutionResponse](#workflowrunnodeexecutionresponse) |
| 403 | Permission denied | |
| 404 | Node last run not found | |
@ -3207,7 +3203,7 @@ Run draft workflow node
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Node run started successfully | [WorkflowRunNodeExecution](#workflowrunnodeexecution) |
| 200 | Node run started successfully | [WorkflowRunNodeExecutionResponse](#workflowrunnodeexecutionresponse) |
| 403 | Permission denied | |
| 404 | Node not found | |
@ -6720,9 +6716,9 @@ Get workflow run list
##### Responses
| Code | Description |
| ---- | ----------- |
| 200 | Success |
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Workflow runs retrieved successfully | [WorkflowRunPaginationResponse](#workflowrunpaginationresponse) |
### /rag/pipelines/{pipeline_id}/workflow-runs/tasks/{task_id}/stop
@ -6760,9 +6756,9 @@ Get workflow run detail
##### Responses
| Code | Description |
| ---- | ----------- |
| 200 | Success |
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Workflow run detail retrieved successfully | [WorkflowRunDetailResponse](#workflowrundetailresponse) |
### /rag/pipelines/{pipeline_id}/workflow-runs/{run_id}/node-executions
@ -6780,9 +6776,9 @@ Get workflow run node execution list
##### Responses
| Code | Description |
| ---- | ----------- |
| 200 | Success |
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Node executions retrieved successfully | [WorkflowRunNodeExecutionListResponse](#workflowrunnodeexecutionlistresponse) |
### /rag/pipelines/{pipeline_id}/workflows
@ -6915,9 +6911,9 @@ Set datasource variables
##### Responses
| Code | Description |
| ---- | ----------- |
| 200 | Success |
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Datasource variables set successfully | [WorkflowRunNodeExecutionResponse](#workflowrunnodeexecutionresponse) |
### /rag/pipelines/{pipeline_id}/workflows/draft/environment-variables
@ -6988,9 +6984,9 @@ Run draft workflow loop node
##### Responses
| Code | Description |
| ---- | ----------- |
| 200 | Success |
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Node last run retrieved successfully | [WorkflowRunNodeExecutionResponse](#workflowrunnodeexecutionresponse) |
### /rag/pipelines/{pipeline_id}/workflows/draft/nodes/{node_id}/run
@ -7009,9 +7005,9 @@ Run draft workflow node
##### Responses
| Code | Description |
| ---- | ----------- |
| 200 | Success |
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Node run started successfully | [WorkflowRunNodeExecutionResponse](#workflowrunnodeexecutionresponse) |
### /rag/pipelines/{pipeline_id}/workflows/draft/nodes/{node_id}/variables
@ -7947,6 +7943,7 @@ Get workflow pause details
##### Description
Get workflow pause details
GET /console/api/workflow/<workflow_run_id>/pause-details
Returns information about why and where the workflow is paused.
@ -7955,13 +7952,14 @@ Returns information about why and where the workflow is paused.
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| workflow_run_id | path | | Yes | string |
| workflow_run_id | path | Workflow run ID | Yes | string |
##### Responses
| Code | Description |
| ---- | ----------- |
| 200 | Success |
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Workflow pause details retrieved successfully | [WorkflowPauseDetailsResponse](#workflowpausedetailsresponse) |
| 404 | Workflow run not found | |
### /workspaces
@ -10256,31 +10254,31 @@ Get banner list
| ---- | ---- | ----------- | -------- |
| result | string | Operation result | Yes |
#### AdvancedChatWorkflowRunForList
#### AdvancedChatWorkflowRunForListResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| conversation_id | string | | No |
| created_at | object | | No |
| created_by_account | [SimpleAccount](#simpleaccount) | | No |
| elapsed_time | number | | No |
| exceptions_count | integer | | No |
| finished_at | object | | No |
| id | string | | No |
| message_id | string | | No |
| retry_index | integer | | No |
| status | string | | No |
| total_steps | integer | | No |
| total_tokens | integer | | No |
| version | string | | No |
| conversation_id | | | No |
| created_at | | | No |
| created_by_account | | | No |
| elapsed_time | | | No |
| exceptions_count | | | No |
| finished_at | | | No |
| id | string | | Yes |
| message_id | | | No |
| retry_index | | | No |
| status | | | No |
| total_steps | | | No |
| total_tokens | | | No |
| version | | | No |
#### AdvancedChatWorkflowRunPagination
#### AdvancedChatWorkflowRunPaginationResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| data | [ [AdvancedChatWorkflowRunForList](#advancedchatworkflowrunforlist) ] | | No |
| has_more | boolean | | No |
| limit | integer | | No |
| data | [ [AdvancedChatWorkflowRunForListResponse](#advancedchatworkflowrunforlistresponse) ] | | Yes |
| has_more | boolean | | Yes |
| limit | integer | | Yes |
#### AdvancedChatWorkflowRunPayload
@ -12169,6 +12167,14 @@ Form input types.
| form_inputs | object | Values the user provides for the form's own fields | Yes |
| inputs | object | Values used to fill missing upstream variables referenced in form_content | Yes |
#### HumanInputPauseTypeResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| backstage_input_url | | | No |
| form_id | string | | Yes |
| type | string | | Yes |
#### IconType
| Name | Type | Description | Required |
@ -13101,6 +13107,14 @@ Enum class for model type.
| ---- | ---- | ----------- | -------- |
| click_id | string | Click Id from partner referral link | Yes |
#### PausedNodeResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| node_id | string | | Yes |
| node_title | string | | Yes |
| pause_type | [HumanInputPauseTypeResponse](#humaninputpausetyperesponse) | | Yes |
#### Payload
| Name | Type | Description | Required |
@ -14306,53 +14320,60 @@ User action configuration.
| updated_at | | | No |
| updated_by | | | No |
#### WorkflowRunCount
#### WorkflowPauseDetailsResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| failed | integer | | No |
| partial_succeeded | integer | | No |
| running | integer | | No |
| stopped | integer | | No |
| succeeded | integer | | No |
| total | integer | | No |
| paused_at | | | No |
| paused_nodes | [ [PausedNodeResponse](#pausednoderesponse) ] | | Yes |
#### WorkflowRunCountQuery
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| status | | Workflow run status filter | No |
| time_range | | Time range filter (e.g., 7d, 4h, 30m, 30s) | No |
| triggered_from | | Filter by trigger source: debugging or app-run | No |
| time_range | | Filter by time range (optional): e.g., 7d (7 days), 4h (4 hours), 30m (30 minutes), 30s (30 seconds). Filters by created_at field. | No |
| triggered_from | | Filter by trigger source: debugging or app-run. Default: debugging | No |
#### WorkflowRunDetail
#### WorkflowRunCountResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| created_at | object | | No |
| created_by_account | [SimpleAccount](#simpleaccount) | | No |
| created_by_end_user | [SimpleEndUser](#simpleenduser) | | No |
| created_by_role | string | | No |
| elapsed_time | number | | No |
| error | string | | No |
| exceptions_count | integer | | No |
| finished_at | object | | No |
| graph | object | | No |
| id | string | | No |
| inputs | object | | No |
| outputs | object | | No |
| status | string | | No |
| total_steps | integer | | No |
| total_tokens | integer | | No |
| version | string | | No |
| failed | integer | | Yes |
| partial_succeeded | integer | | Yes |
| running | integer | | Yes |
| stopped | integer | | Yes |
| succeeded | integer | | Yes |
| total | integer | | Yes |
#### WorkflowRunExport
#### WorkflowRunDetailResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| presigned_url | string | Pre-signed URL for download | No |
| presigned_url_expires_at | string | Pre-signed URL expiration time | No |
| status | string | Export status: success/failed | No |
| created_at | | | No |
| created_by_account | | | No |
| created_by_end_user | | | No |
| created_by_role | | | No |
| elapsed_time | | | No |
| error | | | No |
| exceptions_count | | | No |
| finished_at | | | No |
| graph | | | Yes |
| id | string | | Yes |
| inputs | | | Yes |
| outputs | | | Yes |
| status | | | No |
| total_steps | | | No |
| total_tokens | | | No |
| version | | | No |
#### WorkflowRunExportResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| presigned_url | | Pre-signed URL for download | No |
| presigned_url_expires_at | | Pre-signed URL expiration time | No |
| status | string | Export status: success/failed | Yes |
#### WorkflowRunForArchivedLogResponse
@ -14364,21 +14385,21 @@ User action configuration.
| total_tokens | | | No |
| triggered_from | | | No |
#### WorkflowRunForList
#### WorkflowRunForListResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| created_at | object | | No |
| created_by_account | [SimpleAccount](#simpleaccount) | | No |
| elapsed_time | number | | No |
| exceptions_count | integer | | No |
| finished_at | object | | No |
| id | string | | No |
| retry_index | integer | | No |
| status | string | | No |
| total_steps | integer | | No |
| total_tokens | integer | | No |
| version | string | | No |
| created_at | | | No |
| created_by_account | | | No |
| elapsed_time | | | No |
| exceptions_count | | | No |
| finished_at | | | No |
| id | string | | Yes |
| retry_index | | | No |
| status | | | No |
| total_steps | | | No |
| total_tokens | | | No |
| version | | | No |
#### WorkflowRunForLogResponse
@ -14403,48 +14424,48 @@ User action configuration.
| last_id | | Last run ID for pagination | No |
| limit | integer | Number of items per page (1-100) | No |
| status | | Workflow run status filter | No |
| triggered_from | | Filter by trigger source: debugging or app-run | No |
| triggered_from | | Filter by trigger source: debugging or app-run. Default: debugging | No |
#### WorkflowRunNodeExecution
#### WorkflowRunNodeExecutionListResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| created_at | object | | No |
| created_by_account | [SimpleAccount](#simpleaccount) | | No |
| created_by_end_user | [SimpleEndUser](#simpleenduser) | | No |
| created_by_role | string | | No |
| elapsed_time | number | | No |
| error | string | | No |
| execution_metadata | object | | No |
| extras | object | | No |
| finished_at | object | | No |
| id | string | | No |
| index | integer | | No |
| inputs | object | | No |
| inputs_truncated | boolean | | No |
| node_id | string | | No |
| node_type | string | | No |
| outputs | object | | No |
| outputs_truncated | boolean | | No |
| predecessor_node_id | string | | No |
| process_data | object | | No |
| process_data_truncated | boolean | | No |
| status | string | | No |
| title | string | | No |
| data | [ [WorkflowRunNodeExecutionResponse](#workflowrunnodeexecutionresponse) ] | | Yes |
#### WorkflowRunNodeExecutionList
#### WorkflowRunNodeExecutionResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| data | [ [WorkflowRunNodeExecution](#workflowrunnodeexecution) ] | | No |
| created_at | | | No |
| created_by_account | | | No |
| created_by_end_user | | | No |
| created_by_role | | | No |
| elapsed_time | | | No |
| error | | | No |
| execution_metadata | | | No |
| extras | | | No |
| finished_at | | | No |
| id | string | | Yes |
| index | | | No |
| inputs | | | No |
| inputs_truncated | | | No |
| node_id | | | No |
| node_type | | | No |
| outputs | | | No |
| outputs_truncated | | | No |
| predecessor_node_id | | | No |
| process_data | | | No |
| process_data_truncated | | | No |
| status | | | No |
| title | | | No |
#### WorkflowRunPagination
#### WorkflowRunPaginationResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| data | [ [WorkflowRunForList](#workflowrunforlist) ] | | No |
| has_more | boolean | | No |
| limit | integer | | No |
| data | [ [WorkflowRunForListResponse](#workflowrunforlistresponse) ] | | Yes |
| has_more | boolean | | Yes |
| limit | integer | | Yes |
#### WorkflowRunPayload

41
api/tests/test_containers_integration_tests/controllers/console/datasets/rag_pipeline/test_rag_pipeline_workflow.py
View File

@ -3,6 +3,7 @@
from __future__ import annotations
from datetime import datetime
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
from uuid import uuid4
@ -44,6 +45,35 @@ def unwrap(func):
return func
def make_node_execution(**overrides):
payload = {
"id": "node-exec-1",
"index": 1,
"predecessor_node_id": None,
"node_id": "node1",
"node_type": "start",
"title": "Start",
"inputs_dict": {"query": "hello"},
"process_data_dict": {},
"outputs_dict": {"answer": "world"},
"status": "succeeded",
"error": None,
"elapsed_time": 1.0,
"execution_metadata_dict": {},
"extras": {},
"created_at": datetime(2026, 1, 1, 0, 0, 0),
"created_by_role": "account",
"created_by_account": None,
"created_by_end_user": None,
"finished_at": datetime(2026, 1, 1, 0, 0, 1),
"inputs_truncated": False,
"outputs_truncated": False,
"process_data_truncated": False,
}
payload.update(overrides)
return SimpleNamespace(**payload)
class TestDraftWorkflowApi:
@pytest.fixture
def app(self, flask_app_with_containers: Flask):
@ -743,7 +773,7 @@ class TestRagPipelineWorkflowLastRunApi:
pipeline = MagicMock()
workflow = MagicMock()
node_exec = MagicMock()
node_exec = make_node_execution()
service = MagicMock()
service.get_draft_workflow.return_value = workflow
@ -757,7 +787,9 @@ class TestRagPipelineWorkflowLastRunApi:
),
):
result = method(api, pipeline, "node1")
assert result == node_exec
assert result["id"] == "node-exec-1"
assert result["inputs"] == {"query": "hello"}
assert result["outputs"] == {"answer": "world"}
def test_last_run_not_found(self, app: Flask):
api = RagPipelineWorkflowLastRunApi()
@ -799,7 +831,7 @@ class TestRagPipelineDatasourceVariableApi:
}
service = MagicMock()
service.set_datasource_variables.return_value = MagicMock()
service.set_datasource_variables.return_value = make_node_execution(node_id="n1")
with (
app.test_request_context("/", json=payload),
@ -814,4 +846,5 @@ class TestRagPipelineDatasourceVariableApi:
),
):
result = method(api, pipeline)
assert result is not None
assert result["node_id"] == "n1"
assert result["process_data"] == {}

18
api/tests/unit_tests/controllers/common/test_schema.py
View File

@ -47,6 +47,10 @@ class QueryModel(BaseModel):
ambiguous: int | str | None = Field(default=None, description="Ambiguous query parameter")
class ResponseAliasModel(BaseModel):
public_name: str = Field(validation_alias="internal_name")
@pytest.fixture(autouse=True)
def mock_console_ns():
"""Mock the console_ns to avoid circular imports during test collection."""
@ -146,6 +150,20 @@ def test_register_schema_models_calls_register_schema_model(monkeypatch: pytest.
]
def test_register_response_schema_model_uses_serialized_field_names():
from controllers.common.schema import register_response_schema_model
namespace = MagicMock(spec=Namespace)
register_response_schema_model(namespace, ResponseAliasModel)
model_name, schema = namespace.schema_model.call_args.args
assert model_name == "ResponseAliasModel"
assert "public_name" in schema["properties"]
assert "internal_name" not in schema["properties"]
def test_get_or_create_model_returns_existing_model(mock_console_ns):
from controllers.common.schema import get_or_create_model

21
api/tests/unit_tests/controllers/console/app/test_workflow_pause_details_api.py
View File

@ -112,3 +112,24 @@ def test_pause_details_tenant_isolation(app: Flask, monkeypatch: pytest.MonkeyPa
with pytest.raises(NotFoundError):
with app.test_request_context("/console/api/workflow/run-1/pause-details", method="GET"):
response, status = workflow_run_module.ConsoleWorkflowPauseDetailsApi().get(workflow_run_id="run-1")
def test_pause_details_returns_empty_response_for_non_paused_run(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
account = _make_account()
_patch_console_guards(monkeypatch, account)
workflow_run = Mock(spec=WorkflowRun)
workflow_run.tenant_id = "tenant-123"
workflow_run.status = WorkflowExecutionStatus.RUNNING
fake_db = SimpleNamespace(engine=Mock(), session=SimpleNamespace(get=lambda *_: workflow_run))
monkeypatch.setattr(workflow_run_module, "db", fake_db)
with app.test_request_context("/console/api/workflow/run-1/pause-details", method="GET"):
response, status = workflow_run_module.ConsoleWorkflowPauseDetailsApi().get(workflow_run_id="run-1")
assert status == 200
assert response == {"paused_at": None, "paused_nodes": []}
def test_pause_details_response_schema_is_registered() -> None:
assert workflow_run_module.WorkflowPauseDetailsResponse.__name__ in workflow_run_module.console_ns.models

248
api/tests/unit_tests/controllers/console/app/test_workflow_run_api.py Normal file
View File

@ -0,0 +1,248 @@
from __future__ import annotations
from datetime import UTC, datetime
from types import SimpleNamespace
from typing import Any
import pytest
from flask import Flask
from flask_restx import marshal
from controllers.console.app import workflow_run as workflow_run_module
def _unwrap(func):
while hasattr(func, "__wrapped__"):
func = func.__wrapped__
return func
def _serialize_200_response(handler, payload: Any) -> Any:
response_doc = getattr(handler, "__apidoc__", {}).get("responses", {}).get("200")
if response_doc is None:
return payload
response_model = response_doc[1]
if isinstance(response_model, dict):
return marshal(payload, response_model)
return payload
def _account() -> SimpleNamespace:
return SimpleNamespace(id="account-1", name="Alice", email="alice@example.com")
def _workflow_run_summary(**overrides) -> SimpleNamespace:
created_at = datetime(2026, 1, 2, 3, 4, 5, tzinfo=UTC)
payload = {
"id": "run-1",
"version": "v1",
"status": "succeeded",
"elapsed_time": 1.5,
"total_tokens": 10,
"total_steps": 2,
"created_by_account": _account(),
"created_at": created_at,
"finished_at": created_at,
"exceptions_count": 0,
"retry_index": 0,
}
payload.update(overrides)
return SimpleNamespace(**payload)
def _workflow_run_node_execution(**overrides) -> SimpleNamespace:
created_at = datetime(2026, 1, 2, 3, 4, 5, tzinfo=UTC)
payload = {
"id": "node-exec-1",
"index": 1,
"predecessor_node_id": None,
"node_id": "node-1",
"node_type": "start",
"title": "Start",
"inputs_dict": {"query": "hello"},
"process_data_dict": {"step": "prepared"},
"outputs_dict": {"answer": "world"},
"status": "succeeded",
"error": None,
"elapsed_time": 1.0,
"execution_metadata_dict": {"total_tokens": 3},
"extras": {},
"created_at": created_at,
"created_by_role": "account",
"created_by_account": _account(),
"created_by_end_user": None,
"finished_at": created_at,
"inputs_truncated": False,
"outputs_truncated": False,
"process_data_truncated": False,
}
payload.update(overrides)
return SimpleNamespace(**payload)
def test_workflow_run_list_returns_frontend_history_contract(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
class WorkflowRunService:
def get_paginate_workflow_runs(self, **_kwargs):
return {
"limit": 10,
"has_more": False,
"data": [_workflow_run_summary()],
}
monkeypatch.setattr(workflow_run_module, "WorkflowRunService", WorkflowRunService)
api = workflow_run_module.WorkflowRunListApi()
handler = _unwrap(api.get)
with app.test_request_context("/apps/app-1/workflow-runs?limit=10", method="GET"):
payload = handler(api, app_model=SimpleNamespace(id="app-1", tenant_id="tenant-1"))
response = _serialize_200_response(api.get, payload)
assert response["limit"] == 10
assert response["has_more"] is False
assert response["data"][0] == {
"id": "run-1",
"version": "v1",
"status": "succeeded",
"elapsed_time": 1.5,
"total_tokens": 10,
"total_steps": 2,
"created_by_account": {"id": "account-1", "name": "Alice", "email": "alice@example.com"},
"created_at": 1767323045,
"finished_at": 1767323045,
"exceptions_count": 0,
"retry_index": 0,
}
def test_advanced_chat_workflow_run_list_keeps_message_fields(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
class WorkflowRunService:
def get_paginate_advanced_chat_workflow_runs(self, **_kwargs):
return {
"limit": 1,
"has_more": True,
"data": [
_workflow_run_summary(
conversation_id="conversation-1",
message_id="message-1",
)
],
}
monkeypatch.setattr(workflow_run_module, "WorkflowRunService", WorkflowRunService)
api = workflow_run_module.AdvancedChatAppWorkflowRunListApi()
handler = _unwrap(api.get)
with app.test_request_context("/apps/app-1/advanced-chat/workflow-runs?limit=1", method="GET"):
payload = handler(api, app_model=SimpleNamespace(id="app-1", tenant_id="tenant-1"))
response = _serialize_200_response(api.get, payload)
assert response["data"][0]["conversation_id"] == "conversation-1"
assert response["data"][0]["message_id"] == "message-1"
def test_workflow_run_detail_returns_frontend_detail_contract(app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
created_at = datetime(2026, 1, 2, 3, 4, 5, tzinfo=UTC)
workflow_run = SimpleNamespace(
id="run-1",
version="v1",
graph_dict={"nodes": []},
inputs_dict={"query": "hello"},
status="succeeded",
outputs_dict={"answer": "world"},
error=None,
elapsed_time=1.5,
total_tokens=10,
total_steps=2,
created_by_role="account",
created_by_account=_account(),
created_by_end_user=None,
created_at=created_at,
finished_at=created_at,
exceptions_count=0,
)
class WorkflowRunService:
def get_workflow_run(self, **_kwargs):
return workflow_run
monkeypatch.setattr(workflow_run_module, "WorkflowRunService", WorkflowRunService)
api = workflow_run_module.WorkflowRunDetailApi()
handler = _unwrap(api.get)
with app.test_request_context("/apps/app-1/workflow-runs/run-1", method="GET"):
payload = handler(api, app_model=SimpleNamespace(id="app-1", tenant_id="tenant-1"), run_id="run-1")
response = _serialize_200_response(api.get, payload)
assert response == {
"id": "run-1",
"version": "v1",
"graph": {"nodes": []},
"inputs": {"query": "hello"},
"status": "succeeded",
"outputs": {"answer": "world"},
"error": None,
"elapsed_time": 1.5,
"total_tokens": 10,
"total_steps": 2,
"created_by_role": "account",
"created_by_account": {"id": "account-1", "name": "Alice", "email": "alice@example.com"},
"created_by_end_user": None,
"created_at": 1767323045,
"finished_at": 1767323045,
"exceptions_count": 0,
}
def test_workflow_run_node_executions_return_frontend_trace_contract(
app: Flask, monkeypatch: pytest.MonkeyPatch
) -> None:
class WorkflowRunService:
def get_workflow_run_node_executions(self, **_kwargs):
return [_workflow_run_node_execution()]
monkeypatch.setattr(workflow_run_module, "WorkflowRunService", WorkflowRunService)
monkeypatch.setattr(workflow_run_module, "current_user", SimpleNamespace(id="account-1"))
api = workflow_run_module.WorkflowRunNodeExecutionListApi()
handler = _unwrap(api.get)
with app.test_request_context("/apps/app-1/workflow-runs/run-1/node-executions", method="GET"):
payload = handler(api, app_model=SimpleNamespace(id="app-1", tenant_id="tenant-1"), run_id="run-1")
response = _serialize_200_response(api.get, payload)
assert response == {
"data": [
{
"id": "node-exec-1",
"index": 1,
"predecessor_node_id": None,
"node_id": "node-1",
"node_type": "start",
"title": "Start",
"inputs": {"query": "hello"},
"process_data": {"step": "prepared"},
"outputs": {"answer": "world"},
"status": "succeeded",
"error": None,
"elapsed_time": 1.0,
"execution_metadata": {"total_tokens": 3},
"extras": {},
"created_at": 1767323045,
"created_by_role": "account",
"created_by_account": {"id": "account-1", "name": "Alice", "email": "alice@example.com"},
"created_by_end_user": None,
"finished_at": 1767323045,
"inputs_truncated": False,
"outputs_truncated": False,
"process_data_truncated": False,
}
]
}