opensource/dify
2
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-07-21 02:28:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(api): migrate service app endpoints to BaseModel (#37960)

Browse Source 
Co-authored-by: Asuka Minato <i@asukaminato.eu.org>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
chariri 2026-07-09 13:23:47 +09:00 committed by GitHub
parent 458aa4892d
commit 6540d178c6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
14 changed files with 522 additions and 338 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
api/controllers/service_api/app/annotation.py
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",

8
api/controllers/service_api/app/audio.py
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()

22
api/controllers/service_api/app/completion.py
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

33
api/controllers/service_api/app/conversation.py
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:

4
api/controllers/service_api/app/file.py
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

35
api/controllers/service_api/app/message.py
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")

104
api/controllers/service_api/app/workflow.py
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)

69
api/openapi/markdown/service-openapi.md
View File

@ -167,7 +167,7 @@ Retrieves the status of an asynchronous annotation reply configuration job start
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Successfully retrieved task status. | **application/json**: [AnnotationJobStatusResponse](#annotationjobstatusresponse)<br> |
| 200 | Successfully retrieved task status. | **application/json**: [AnnotationJobStatusDetailResponse](#annotationjobstatusdetailresponse)<br> |
| 400 | `invalid_param` : The specified job does not exist. | |
| 401 | Unauthorized - invalid API token | |
| 403 | Forbidden - token scope, app, dataset, or workspace access denied | |
@ -322,15 +322,15 @@ Send a request to the chat application.
#### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Successful response. The content type and structure depend on the `response_mode` parameter in the request. - If `response_mode` is `blocking`, returns `application/json` with a `ChatCompletionResponse` object. - If `response_mode` is `streaming`, returns `text/event-stream` with a stream of Server-Sent Events. | **application/json**: [GeneratedAppResponse](#generatedappresponse)<br>**text/event-stream**: [GeneratedAppResponse](#generatedappresponse)<br> |
| 400 | - `app_unavailable` : App unavailable or misconfigured. - `not_chat_app` : App mode does not match the API route. - `conversation_completed` : The conversation has ended. - `provider_not_initialize` : No valid model provider credentials found. - `provider_quota_exceeded` : Model provider quota exhausted. - `model_currently_not_support` : Current model unavailable. - `completion_request_error` : Text generation failed. | |
| 401 | Unauthorized - invalid API token | |
| 403 | Forbidden - token scope, app, dataset, or workspace access denied | |
| 404 | `not_found` : Conversation does not exist. | |
| 429 | - `too_many_requests` : Too many concurrent requests for this app. - `rate_limit_error` : The upstream model provider rate limit was exceeded. | |
| 500 | `internal_server_error` : Internal server error. | |
| Code | Description |
| ---- | ----------- |
| 200 | Successful response. The content type and structure depend on the `response_mode` parameter in the request. - If `response_mode` is `blocking`, returns `application/json` with a `ChatCompletionResponse` object. - If `response_mode` is `streaming`, returns `text/event-stream` with a stream of Server-Sent Events. |
| 400 | - `app_unavailable` : App unavailable or misconfigured. - `not_chat_app` : App mode does not match the API route. - `conversation_completed` : The conversation has ended. - `provider_not_initialize` : No valid model provider credentials found. - `provider_quota_exceeded` : Model provider quota exhausted. - `model_currently_not_support` : Current model unavailable. - `completion_request_error` : Text generation failed. |
| 401 | Unauthorized - invalid API token |
| 403 | Forbidden - token scope, app, dataset, or workspace access denied |
| 404 | `not_found` : Conversation does not exist. |
| 429 | - `too_many_requests` : Too many concurrent requests for this app. - `rate_limit_error` : The upstream model provider rate limit was exceeded. |
| 500 | `internal_server_error` : Internal server error. |
### [POST] /chat-messages/{task_id}/stop
**Stop Chat Message Generation**
@ -469,15 +469,15 @@ Send a request to the chat application.
#### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Successful response. The content type and structure depend on the `response_mode` parameter in the request. - If `response_mode` is `blocking`, returns `application/json` with a `ChatCompletionResponse` object. - If `response_mode` is `streaming`, returns `text/event-stream` with a stream of Server-Sent Events. | **application/json**: [GeneratedAppResponse](#generatedappresponse)<br>**text/event-stream**: [GeneratedAppResponse](#generatedappresponse)<br> |
| 400 | - `app_unavailable` : App unavailable or misconfigured. - `not_chat_app` : App mode does not match the API route. - `conversation_completed` : The conversation has ended. - `provider_not_initialize` : No valid model provider credentials found. - `provider_quota_exceeded` : Model provider quota exhausted. - `model_currently_not_support` : Current model unavailable. - `completion_request_error` : Text generation failed. | |
| 401 | Unauthorized - invalid API token | |
| 403 | Forbidden - token scope, app, dataset, or workspace access denied | |
| 404 | `not_found` : Conversation does not exist. | |
| 429 | - `too_many_requests` : Too many concurrent requests for this app. - `rate_limit_error` : The upstream model provider rate limit was exceeded. | |
| 500 | `internal_server_error` : Internal server error. | |
| Code | Description |
| ---- | ----------- |
| 200 | Successful response. The content type and structure depend on the `response_mode` parameter in the request. - If `response_mode` is `blocking`, returns `application/json` with a `ChatCompletionResponse` object. - If `response_mode` is `streaming`, returns `text/event-stream` with a stream of Server-Sent Events. |
| 400 | - `app_unavailable` : App unavailable or misconfigured. - `not_chat_app` : App mode does not match the API route. - `conversation_completed` : The conversation has ended. - `provider_not_initialize` : No valid model provider credentials found. - `provider_quota_exceeded` : Model provider quota exhausted. - `model_currently_not_support` : Current model unavailable. - `completion_request_error` : Text generation failed. |
| 401 | Unauthorized - invalid API token |
| 403 | Forbidden - token scope, app, dataset, or workspace access denied |
| 404 | `not_found` : Conversation does not exist. |
| 429 | - `too_many_requests` : Too many concurrent requests for this app. - `rate_limit_error` : The upstream model provider rate limit was exceeded. |
| 500 | `internal_server_error` : Internal server error. |
### [POST] /chat-messages/{task_id}/stop
**Stop Chat Message Generation**
@ -545,15 +545,15 @@ Send a request to the text generation application.
#### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Successful response. The content type and structure depend on the `response_mode` parameter in the request. - If `response_mode` is `blocking`, returns `application/json` with a `CompletionResponse` object. - If `response_mode` is `streaming`, returns `text/event-stream` with a stream of `ChunkCompletionEvent` objects. | **application/json**: [GeneratedAppResponse](#generatedappresponse)<br>**text/event-stream**: [GeneratedAppResponse](#generatedappresponse)<br> |
| 400 | - `app_unavailable` : App unavailable or misconfigured. - `provider_not_initialize` : No valid model provider credentials found. - `provider_quota_exceeded` : Model provider quota exhausted. - `model_currently_not_support` : Current model unavailable. - `completion_request_error` : Text generation failed. | |
| 401 | Unauthorized - invalid API token | |
| 403 | Forbidden - token scope, app, dataset, or workspace access denied | |
| 404 | Conversation not found | |
| 429 | `too_many_requests` : Too many concurrent requests for this app. | |
| 500 | `internal_server_error` : Internal server error. | |
| Code | Description |
| ---- | ----------- |
| 200 | Successful response. The content type and structure depend on the `response_mode` parameter in the request. - If `response_mode` is `blocking`, returns `application/json` with a `CompletionResponse` object. - If `response_mode` is `streaming`, returns `text/event-stream` with a stream of `ChunkCompletionEvent` objects. |
| 400 | - `app_unavailable` : App unavailable or misconfigured. - `provider_not_initialize` : No valid model provider credentials found. - `provider_quota_exceeded` : Model provider quota exhausted. - `model_currently_not_support` : Current model unavailable. - `completion_request_error` : Text generation failed. |
| 401 | Unauthorized - invalid API token |
| 403 | Forbidden - token scope, app, dataset, or workspace access denied |
| 404 | Conversation not found |
| 429 | `too_many_requests` : Too many concurrent requests for this app. |
| 500 | `internal_server_error` : Internal server error. |
### [POST] /completion-messages/{task_id}/stop
**Stop Completion Message Generation**
@ -2285,7 +2285,7 @@ Retrieve the list of available models by type. Primarily used to query `text-emb
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| content | string | | No |
| answer | string | | No |
| created_at | integer | | No |
| hit_count | integer | | No |
| id | string | | Yes |
@ -2298,13 +2298,20 @@ Retrieve the list of available models by type. Primarily used to query `text-emb
| answer | string | Annotation answer. | Yes |
| question | string | Annotation question. | Yes |
#### AnnotationJobStatusResponse
#### AnnotationJobStatusDetailResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| error_msg | string | | No |
| job_id | string | | Yes |
| job_status | string | | Yes |
| job_status | string<br>string | | Yes |
#### AnnotationJobStatusResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| job_id | string | | Yes |
| job_status | string<br>string | | Yes |
#### AnnotationList

10
api/tests/unit_tests/controllers/service_api/app/test_annotation.py
View File

@ -141,14 +141,14 @@ class TestAppModelPatterns:
assert app.id is not None
assert app.status == "normal"
assert app.enable_api is True
assert app.enable_api
def test_app_model_disabled_api(self):
"""Test app with disabled API access."""
app = Mock(spec=App)
app.enable_api = False
assert app.enable_api is False
assert not app.enable_api
def test_app_model_archived_status(self):
"""Test app with archived status."""
@ -183,7 +183,7 @@ class TestAnnotationErrorPatterns:
class TestAnnotationReplyActionApi:
def test_enable(self, app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
enable_mock = Mock()
enable_mock = Mock(return_value={"job_id": "job-1", "job_status": "waiting"})
monkeypatch.setattr(AppAnnotationService, "enable_app_annotation", enable_mock)
api = AnnotationReplyActionApi()
@ -198,10 +198,11 @@ class TestAnnotationReplyActionApi:
response, status = handler(api, app_model=app_model, action="enable")
assert status == 200
assert response == {"job_id": "job-1", "job_status": "waiting"}
enable_mock.assert_called_once()
def test_disable(self, app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
disable_mock = Mock()
disable_mock = Mock(return_value={"job_id": "job-1", "job_status": "waiting"})
monkeypatch.setattr(AppAnnotationService, "disable_app_annotation", disable_mock)
api = AnnotationReplyActionApi()
@ -216,6 +217,7 @@ class TestAnnotationReplyActionApi:
response, status = handler(api, app_model=app_model, action="disable")
assert status == 200
assert response == {"job_id": "job-1", "job_status": "waiting"}
disable_mock.assert_called_once()

30
api/tests/unit_tests/controllers/service_api/app/test_message.py
View File

@ -51,7 +51,7 @@ class TestMessageListQuery:
"""Test conversation_id is required."""
conversation_id = str(uuid.uuid4())
query = MessageListQuery(conversation_id=conversation_id)
assert str(query.conversation_id) == conversation_id
assert query.conversation_id == conversation_id
def test_query_with_defaults(self):
"""Test query with default values."""
@ -87,13 +87,13 @@ class TestMessageListQuery:
"""Test query rejects limit < 1."""
conversation_id = str(uuid.uuid4())
with pytest.raises(ValueError):
MessageListQuery(conversation_id=conversation_id, limit=0)
MessageListQuery(conversation_id=conversation_id, limit=0) # pyrefly: ignore[bad-argument-type]
def test_query_rejects_limit_above_maximum(self):
"""Test query rejects limit > 100."""
conversation_id = str(uuid.uuid4())
with pytest.raises(ValueError):
MessageListQuery(conversation_id=conversation_id, limit=101)
MessageListQuery(conversation_id=conversation_id, limit=101) # pyrefly: ignore[bad-argument-type]
class TestMessageFeedbackPayload:
@ -131,6 +131,7 @@ class TestMessageFeedbackPayload:
"""Test payload with long feedback content."""
long_content = "A" * 1000
payload = MessageFeedbackPayload(content=long_content)
assert payload.content is not None
assert len(payload.content) == 1000
def test_payload_with_unicode_content(self):
@ -163,7 +164,7 @@ class TestFeedbackListQuery:
def test_query_rejects_page_below_minimum(self):
"""Test query rejects page < 1."""
with pytest.raises(ValueError):
FeedbackListQuery(page=0)
FeedbackListQuery(page=0) # pyrefly: ignore[bad-argument-type]
def test_query_limit_boundaries(self):
"""Test query limit boundaries."""
@ -176,12 +177,12 @@ class TestFeedbackListQuery:
def test_query_rejects_limit_below_minimum(self):
"""Test query rejects limit < 1."""
with pytest.raises(ValueError):
FeedbackListQuery(limit=0)
FeedbackListQuery(limit=0) # pyrefly: ignore[bad-argument-type]
def test_query_rejects_limit_above_maximum(self):
"""Test query rejects limit > 101."""
with pytest.raises(ValueError):
FeedbackListQuery(limit=102)
FeedbackListQuery(limit=102) # pyrefly: ignore[bad-argument-type]
class TestMessageAppModeValidation:
@ -470,7 +471,20 @@ class TestMessageFeedbackApi:
class TestAppGetFeedbacksApi:
def test_success(self, app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(MessageService, "get_all_messages_feedbacks", lambda *_args, **_kwargs: ["f1"])
feedback = {
"id": "feedback-1",
"app_id": "app-1",
"conversation_id": "conversation-1",
"message_id": "message-1",
"rating": "like",
"content": "helpful answer",
"from_source": "user",
"from_end_user_id": "end-user-1",
"from_account_id": None,
"created_at": "2024-01-02T03:04:05",
"updated_at": "2024-01-02T03:04:06",
}
monkeypatch.setattr(MessageService, "get_all_messages_feedbacks", lambda *_args, **_kwargs: [feedback])
api = AppGetFeedbacksApi()
handler = unwrap(api.get)
@ -479,7 +493,7 @@ class TestAppGetFeedbacksApi:
with app.test_request_context("/app/feedbacks?page=1&limit=20", method="GET"):
response = handler(api, app_model=app_model)
assert response == {"data": ["f1"]}
assert response == {"data": [feedback]}
class TestMessageSuggestedApi:

412
api/tests/unit_tests/controllers/service_api/app/test_workflow.py
View File

@ -13,15 +13,17 @@ Focus on:
- Service method interfaces
"""
import json
import sys
import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime
from inspect import unwrap
from types import SimpleNamespace
from unittest.mock import MagicMock, Mock, patch
import pytest
from flask import Flask
from sqlalchemy.orm import sessionmaker
from werkzeug.exceptions import BadRequest, NotFound
from controllers.service_api.app.error import NotWorkflowAppError
@ -35,31 +37,175 @@ from controllers.service_api.app.workflow import (
WorkflowRunByIdApi,
WorkflowRunDetailApi,
WorkflowRunPayload,
WorkflowRunResponse,
WorkflowTaskStopApi,
)
from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
from core.app.entities.app_invoke_entities import InvokeFrom
from graphon.enums import WorkflowExecutionStatus
from models.model import App, AppMode
from models.enums import CreatorUserRole, WorkflowRunTriggeredFrom
from models.model import App, AppMode, EndUser
from models.workflow import WorkflowAppLog, WorkflowAppLogCreatedFrom, WorkflowRun, WorkflowType
from services.app_generate_service import AppGenerateService
from services.errors.app import IsDraftWorkflowError, WorkflowNotFoundError
from services.errors.llm import InvokeRateLimitError
from services.workflow_app_service import WorkflowAppService
from services.workflow_app_service import LogView, LogViewDetails, WorkflowAppService
def _make_mock_workflow_run(run_id: str = "run-1"):
run = Mock()
run.id = run_id
run.workflow_id = "wf-1"
run.status = WorkflowExecutionStatus.SUCCEEDED
run.inputs = {"input": "value"}
run.outputs_dict = {"output": "value"}
run.error = None
run.total_steps = 1
run.total_tokens = 10
run.created_at = datetime(2026, 1, 1, tzinfo=UTC)
run.finished_at = datetime(2026, 1, 1, tzinfo=UTC)
run.elapsed_time = 0.1
return run
def _default_workflow_inputs() -> dict[str, object]:
return {"input": "value"}
def _default_log_details() -> LogViewDetails:
return {"trigger_metadata": {"node": "answer", "latency": 1.25}}
class _DbSessionStub:
def get(self, *args: object, **kwargs: object) -> None:
return None
@dataclass
class _DbStub:
engine: object = field(default_factory=object)
session: _DbSessionStub = field(default_factory=_DbSessionStub)
@dataclass
class _WorkflowRunRepositoryStub:
run: WorkflowRun | None
def get_workflow_run_by_id(self, *, tenant_id: str, app_id: str, run_id: str) -> WorkflowRun | None:
return self.run if tenant_id and app_id and run_id else None
def get_workflow_run_by_id_without_tenant(self, *, run_id: str) -> WorkflowRun | None:
return self.run if run_id else None
class _BeginStub:
def __enter__(self) -> object:
return object()
def __exit__(self, exc_type: object, exc: object, tb: object) -> bool:
return False
class _SessionMakerStub:
def __init__(self, *args: object, **kwargs: object) -> None:
pass
def begin(self) -> _BeginStub:
return _BeginStub()
def _make_workflow_run(
run_id: str = "run-1",
*,
workflow_id: str = "wf-1",
inputs: dict[str, object] | None = None,
outputs: dict[str, object] | None = None,
created_at: datetime | None = None,
finished_at: datetime | None = None,
) -> WorkflowRun:
return WorkflowRun(
id=run_id,
tenant_id="tenant-1",
app_id="app-1",
workflow_id=workflow_id,
type=WorkflowType.WORKFLOW,
triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
version="2026-01-01",
graph=json.dumps({"nodes": [], "edges": []}),
inputs=json.dumps(inputs if inputs is not None else _default_workflow_inputs()),
outputs=json.dumps(outputs if outputs is not None else {"output": "value"}),
status=WorkflowExecutionStatus.SUCCEEDED,
error=None,
elapsed_time=0.1,
total_tokens=10,
total_steps=1,
created_by_role=CreatorUserRole.END_USER,
created_by="end-user-1",
created_at=created_at or datetime(2026, 1, 1, tzinfo=UTC),
finished_at=finished_at or datetime(2026, 1, 1, tzinfo=UTC),
exceptions_count=0,
)
def _make_workflow_app_log() -> WorkflowAppLog:
log = WorkflowAppLog(
tenant_id="tenant-1",
app_id="app-1",
workflow_id="wf-1",
workflow_run_id="log-run-1",
created_from=WorkflowAppLogCreatedFrom.SERVICE_API,
created_by_role=CreatorUserRole.ACCOUNT,
created_by="account-1",
)
log.id = "app-log-1"
log.created_at = datetime(2026, 1, 1, 1, 0, 3, tzinfo=UTC)
return log
def _make_workflow_log_page() -> dict[str, object]:
return {
"page": 1,
"limit": 20,
"total": 1,
"has_more": False,
"data": [LogView(_make_workflow_app_log(), _default_log_details())],
}
def _make_app_model(
*,
app_id: str = "app-1",
tenant_id: str = "tenant-1",
mode: AppMode = AppMode.WORKFLOW,
) -> App:
app = App()
app.id = app_id
app.tenant_id = tenant_id
app.mode = mode
return app
def _make_end_user(user_id: str = "end-user-1") -> EndUser:
end_user = EndUser()
end_user.id = user_id
return end_user
def _expected_workflow_log_pagination_payload() -> dict[str, object]:
return {
"page": 1,
"limit": 20,
"total": 1,
"has_more": False,
"data": [
{
"id": "app-log-1",
"workflow_run": {
"id": "log-run-1",
"version": "2026-01-01",
"status": "succeeded",
"triggered_from": "app-run",
"error": None,
"elapsed_time": 0.1,
"total_tokens": 10,
"total_steps": 1,
"created_at": 1767229200,
"finished_at": 1767229202,
"exceptions_count": 0,
},
"details": {"trigger_metadata": {"node": "answer", "latency": 1.25}},
"created_from": "service-api",
"created_by_role": "account",
"created_by_account": None,
"created_by_end_user": None,
"created_at": 1767229203,
}
],
}
class TestWorkflowRunPayload:
@ -108,6 +254,7 @@ class TestWorkflowRunPayload:
{"type": "audio", "url": "http://example.com/audio.mp3"},
]
payload = WorkflowRunPayload(inputs={}, files=files)
assert payload.files is not None
assert len(payload.files) == 3
@ -179,6 +326,29 @@ class TestWorkflowLogQuery:
assert query.created_at__after == "2024-01-01T00:00:00Z"
class TestWorkflowRunResponse:
def test_validates_workflow_run_object_shape_and_clears_paused_outputs(self):
run = _make_workflow_run(run_id="run-paused")
run.status = WorkflowExecutionStatus.PAUSED
run.outputs = json.dumps({"should": "not leak"})
result = WorkflowRunResponse.model_validate(run, from_attributes=True).model_dump(mode="json")
assert result == {
"id": "run-paused",
"workflow_id": "wf-1",
"status": "paused",
"inputs": '{"input": "value"}',
"outputs": {},
"error": None,
"total_steps": 1,
"total_tokens": 10,
"created_at": 1767225600,
"finished_at": 1767225600,
"elapsed_time": 0.1,
}
class TestWorkflowAppService:
"""Test WorkflowAppService interface."""
@ -195,17 +365,13 @@ class TestWorkflowAppService:
@patch.object(WorkflowAppService, "get_paginate_workflow_app_logs")
def test_get_paginate_workflow_app_logs_returns_pagination(self, mock_get_logs):
"""Test get_paginate_workflow_app_logs returns paginated result."""
mock_pagination = Mock()
mock_pagination.data = []
mock_pagination.page = 1
mock_pagination.limit = 20
mock_pagination.total = 0
mock_get_logs.return_value = mock_pagination
pagination = _make_workflow_log_page()
mock_get_logs.return_value = pagination
service = WorkflowAppService()
result = service.get_paginate_workflow_app_logs(
session=Mock(),
app_model=Mock(spec=App),
app_model=_make_app_model(),
keyword=None,
status=None,
created_at_before=None,
@ -216,8 +382,7 @@ class TestWorkflowAppService:
created_by_account=None,
)
assert result.page == 1
assert result.limit == 20
assert result == pagination
class TestWorkflowExecutionStatus:
@ -248,10 +413,10 @@ class TestAppGenerateServiceWorkflow:
mock_generate.return_value = {"result": "success"}
result = AppGenerateService.generate(
app_model=Mock(spec=App),
user=Mock(),
app_model=_make_app_model(),
user=_make_end_user(),
args={"inputs": {"key": "value"}, "workflow_id": "workflow_123"},
invoke_from=Mock(),
invoke_from=InvokeFrom.SERVICE_API,
session=MagicMock(),
streaming=False,
)
@ -266,10 +431,10 @@ class TestAppGenerateServiceWorkflow:
with pytest.raises(WorkflowNotFoundError):
AppGenerateService.generate(
app_model=Mock(spec=App),
user=Mock(),
app_model=_make_app_model(),
user=_make_end_user(),
args={"workflow_id": "invalid_id"},
invoke_from=Mock(),
invoke_from=InvokeFrom.SERVICE_API,
session=MagicMock(),
streaming=False,
)
@ -281,10 +446,10 @@ class TestAppGenerateServiceWorkflow:
with pytest.raises(IsDraftWorkflowError):
AppGenerateService.generate(
app_model=Mock(spec=App),
user=Mock(),
app_model=_make_app_model(),
user=_make_end_user(),
args={"workflow_id": "draft_workflow"},
invoke_from=Mock(),
invoke_from=InvokeFrom.SERVICE_API,
session=MagicMock(),
streaming=False,
)
@ -296,10 +461,10 @@ class TestAppGenerateServiceWorkflow:
mock_generate.return_value = mock_stream
result = AppGenerateService.generate(
app_model=Mock(spec=App),
user=Mock(),
app_model=_make_app_model(),
user=_make_end_user(),
args={"inputs": {}, "response_mode": "streaming"},
invoke_from=Mock(),
invoke_from=InvokeFrom.SERVICE_API,
session=MagicMock(),
streaming=True,
)
@ -335,37 +500,33 @@ class TestWorkflowRunRepository:
@patch("repositories.factory.DifyAPIRepositoryFactory.create_api_workflow_run_repository")
def test_workflow_run_repository_get_by_id(self, mock_factory):
"""Test workflow run repository get_workflow_run_by_id method."""
mock_repo = Mock()
mock_run = Mock()
mock_run.id = str(uuid.uuid4())
mock_run.status = "succeeded"
mock_repo.get_workflow_run_by_id.return_value = mock_run
mock_factory.return_value = mock_repo
run = _make_workflow_run(run_id=str(uuid.uuid4()))
mock_factory.return_value = _WorkflowRunRepositoryStub(run=run)
from repositories.factory import DifyAPIRepositoryFactory
repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(Mock())
repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(sessionmaker())
result = repo.get_workflow_run_by_id(tenant_id="tenant_123", app_id="app_456", run_id="run_789")
assert result.status == "succeeded"
assert result == run
class TestWorkflowRunDetailApi:
def test_not_workflow_app(self, app: Flask) -> None:
api = WorkflowRunDetailApi()
handler = unwrap(api.get)
app_model = SimpleNamespace(mode=AppMode.CHAT.value)
app_model = _make_app_model(mode=AppMode.CHAT)
with app.test_request_context("/workflows/run/1", method="GET"):
with pytest.raises(NotWorkflowAppError):
handler(api, app_model=app_model, workflow_run_id="run")
def test_success(self, monkeypatch: pytest.MonkeyPatch) -> None:
run = _make_mock_workflow_run(run_id="run")
repo = SimpleNamespace(get_workflow_run_by_id=lambda **_kwargs: run)
run = _make_workflow_run(run_id="run")
repo = _WorkflowRunRepositoryStub(run=run)
workflow_module = sys.modules["controllers.service_api.app.workflow"]
monkeypatch.setattr(workflow_module, "db", SimpleNamespace(engine=object()))
monkeypatch.setattr(workflow_module, "db", _DbStub())
monkeypatch.setattr(
DifyAPIRepositoryFactory,
"create_api_workflow_run_repository",
@ -374,7 +535,7 @@ class TestWorkflowRunDetailApi:
api = WorkflowRunDetailApi()
handler = unwrap(api.get)
app_model = SimpleNamespace(mode=AppMode.WORKFLOW, tenant_id="t1", id="a1")
app_model = _make_app_model(app_id="a1", tenant_id="t1")
result = handler(api, app_model=app_model, workflow_run_id="run")
assert result["id"] == "run"
@ -386,8 +547,8 @@ class TestWorkflowRunApi:
def test_not_workflow_app(self, app: Flask) -> None:
api = WorkflowRunApi()
handler = unwrap(api.post)
app_model = SimpleNamespace(mode=AppMode.CHAT.value)
end_user = SimpleNamespace()
app_model = _make_app_model(mode=AppMode.CHAT)
end_user = _make_end_user()
with app.test_request_context("/workflows/run", method="POST", json={"inputs": {}}):
with pytest.raises(NotWorkflowAppError):
@ -402,8 +563,8 @@ class TestWorkflowRunApi:
api = WorkflowRunApi()
handler = unwrap(api.post)
app_model = SimpleNamespace(mode=AppMode.WORKFLOW)
end_user = SimpleNamespace()
app_model = _make_app_model()
end_user = _make_end_user()
with app.test_request_context("/workflows/run", method="POST", json={"inputs": {}}):
with pytest.raises(InvokeRateLimitHttpError):
@ -420,8 +581,8 @@ class TestWorkflowRunByIdApi:
api = WorkflowRunByIdApi()
handler = unwrap(api.post)
app_model = SimpleNamespace(mode=AppMode.WORKFLOW)
end_user = SimpleNamespace()
app_model = _make_app_model()
end_user = _make_end_user()
with app.test_request_context("/workflows/1/run", method="POST", json={"inputs": {}}):
with pytest.raises(NotFound):
@ -436,8 +597,8 @@ class TestWorkflowRunByIdApi:
api = WorkflowRunByIdApi()
handler = unwrap(api.post)
app_model = SimpleNamespace(mode=AppMode.WORKFLOW)
end_user = SimpleNamespace()
app_model = _make_app_model()
end_user = _make_end_user()
with app.test_request_context("/workflows/1/run", method="POST", json={"inputs": {}}):
with pytest.raises(BadRequest):
@ -448,8 +609,8 @@ class TestWorkflowTaskStopApi:
def test_wrong_mode(self, app: Flask) -> None:
api = WorkflowTaskStopApi()
handler = unwrap(api.post)
app_model = SimpleNamespace(mode=AppMode.CHAT.value)
end_user = SimpleNamespace()
app_model = _make_app_model(mode=AppMode.CHAT)
end_user = _make_end_user()
with app.test_request_context("/workflows/tasks/1/stop", method="POST"):
with pytest.raises(NotWorkflowAppError):
@ -463,8 +624,8 @@ class TestWorkflowTaskStopApi:
api = WorkflowTaskStopApi()
handler = unwrap(api.post)
app_model = SimpleNamespace(mode=AppMode.WORKFLOW)
end_user = SimpleNamespace(id="u1")
app_model = _make_app_model()
end_user = _make_end_user(user_id="u1")
with app.test_request_context("/workflows/tasks/1/stop", method="POST"):
response = handler(api, app_model=app_model, end_user=end_user, task_id="t1")
@ -476,37 +637,36 @@ class TestWorkflowTaskStopApi:
class TestWorkflowAppLogApi:
def test_success(self, app: Flask, monkeypatch: pytest.MonkeyPatch) -> None:
class _BeginStub:
def __enter__(self):
return SimpleNamespace()
def __exit__(self, exc_type, exc, tb):
return False
class _SessionMakerStub:
def __init__(self, *args, **kwargs):
pass
def begin(self):
return _BeginStub()
workflow_module = sys.modules["controllers.service_api.app.workflow"]
monkeypatch.setattr(workflow_module, "db", SimpleNamespace(engine=object()))
workflow_model_module = sys.modules["models.workflow"]
monkeypatch.setattr(workflow_module, "db", _DbStub())
monkeypatch.setattr(workflow_model_module, "db", _DbStub())
monkeypatch.setattr(workflow_module, "sessionmaker", _SessionMakerStub)
monkeypatch.setattr(
WorkflowAppService,
"get_paginate_workflow_app_logs",
lambda *_args, **_kwargs: {"page": 1, "limit": 20, "total": 0, "has_more": False, "data": []},
lambda *_args, **_kwargs: _make_workflow_log_page(),
)
monkeypatch.setattr(
DifyAPIRepositoryFactory,
"create_api_workflow_run_repository",
lambda *_args, **_kwargs: _WorkflowRunRepositoryStub(
run=_make_workflow_run(
run_id="log-run-1",
created_at=datetime(2026, 1, 1, 1, tzinfo=UTC),
finished_at=datetime(2026, 1, 1, 1, 0, 2, tzinfo=UTC),
)
),
)
api = WorkflowAppLogApi()
handler = unwrap(api.get)
app_model = SimpleNamespace(id="a1")
app_model = _make_app_model(app_id="a1")
with app.test_request_context("/workflows/logs", method="GET"):
response = handler(api, app_model=app_model)
assert response == {"page": 1, "limit": 20, "total": 0, "has_more": False, "data": []}
assert response == _expected_workflow_log_pagination_payload()
# =============================================================================
@ -520,12 +680,8 @@ class TestWorkflowAppLogApi:
@pytest.fixture
def mock_workflow_app():
app = Mock(spec=App)
app.id = str(uuid.uuid4())
app.tenant_id = str(uuid.uuid4())
app.mode = AppMode.WORKFLOW
return app
def workflow_app() -> App:
return _make_app_model(app_id=str(uuid.uuid4()), tenant_id=str(uuid.uuid4()))
class TestWorkflowRunDetailApiGet:
@ -542,38 +698,46 @@ class TestWorkflowRunDetailApiGet:
mock_db,
mock_repo_factory,
app: Flask,
mock_workflow_app,
workflow_app: App,
):
"""Test successful workflow run detail retrieval."""
mock_run = _make_mock_workflow_run(run_id="run-1")
mock_repo = Mock()
mock_repo.get_workflow_run_by_id.return_value = mock_run
mock_repo_factory.create_api_workflow_run_repository.return_value = mock_repo
run = _make_workflow_run(run_id="run-1")
mock_repo_factory.create_api_workflow_run_repository.return_value = _WorkflowRunRepositoryStub(run=run)
from controllers.service_api.app.workflow import WorkflowRunDetailApi
with app.test_request_context(
f"/workflows/run/{mock_run.id}",
f"/workflows/run/{run.id}",
method="GET",
):
api = WorkflowRunDetailApi()
result = unwrap(api.get)(api, app_model=mock_workflow_app, workflow_run_id=mock_run.id)
result = unwrap(api.get)(api, app_model=workflow_app, workflow_run_id=run.id)
assert result["id"] == mock_run.id
assert result["status"] == "succeeded"
assert result == {
"id": "run-1",
"workflow_id": "wf-1",
"status": "succeeded",
"inputs": '{"input": "value"}',
"outputs": {"output": "value"},
"error": None,
"total_steps": 1,
"total_tokens": 10,
"created_at": 1767225600,
"finished_at": 1767225600,
"elapsed_time": 0.1,
}
@patch("controllers.service_api.app.workflow.db")
def test_get_workflow_run_wrong_app_mode(self, mock_db, app: Flask):
"""Test NotWorkflowAppError when app mode is not workflow or advanced_chat."""
from controllers.service_api.app.workflow import WorkflowRunDetailApi
mock_app = Mock(spec=App)
mock_app.mode = AppMode.CHAT.value
app_model = _make_app_model(mode=AppMode.CHAT)
with app.test_request_context("/workflows/run/run-1", method="GET"):
api = WorkflowRunDetailApi()
with pytest.raises(NotWorkflowAppError):
unwrap(api.get)(api, app_model=mock_app, workflow_run_id="run-1")
unwrap(api.get)(api, app_model=app_model, workflow_run_id="run-1")
class TestWorkflowTaskStopApiPost:
@ -589,7 +753,7 @@ class TestWorkflowTaskStopApiPost:
mock_queue_mgr,
mock_graph_mgr,
app: Flask,
mock_workflow_app,
workflow_app: App,
):
"""Test successful workflow task stop."""
from controllers.service_api.app.workflow import WorkflowTaskStopApi
@ -598,8 +762,8 @@ class TestWorkflowTaskStopApiPost:
api = WorkflowTaskStopApi()
result = unwrap(api.post)(
api,
app_model=mock_workflow_app,
end_user=Mock(),
app_model=workflow_app,
end_user=_make_end_user(),
task_id="task-1",
)
@ -612,13 +776,12 @@ class TestWorkflowTaskStopApiPost:
"""Test NotWorkflowAppError when app mode is not workflow."""
from controllers.service_api.app.workflow import WorkflowTaskStopApi
mock_app = Mock(spec=App)
mock_app.mode = AppMode.COMPLETION.value
app_model = _make_app_model(mode=AppMode.COMPLETION)
with app.test_request_context("/workflows/tasks/task-1/stop", method="POST"):
api = WorkflowTaskStopApi()
with pytest.raises(NotWorkflowAppError):
unwrap(api.post)(api, app_model=mock_app, end_user=Mock(), task_id="task-1")
unwrap(api.post)(api, app_model=app_model, end_user=_make_end_user(), task_id="task-1")
class TestWorkflowAppLogApiGet:
@ -634,27 +797,23 @@ class TestWorkflowAppLogApiGet:
mock_db,
mock_wf_svc_cls,
app: Flask,
mock_workflow_app,
workflow_app: App,
):
"""Test successful workflow log retrieval."""
mock_pagination = Mock()
mock_pagination.page = 1
mock_pagination.limit = 20
mock_pagination.total = 0
mock_pagination.has_more = False
mock_pagination.data = []
mock_svc_instance = Mock()
mock_svc_instance.get_paginate_workflow_app_logs.return_value = mock_pagination
mock_svc_instance.get_paginate_workflow_app_logs.return_value = _make_workflow_log_page()
mock_wf_svc_cls.return_value = mock_svc_instance
mock_repo = _WorkflowRunRepositoryStub(
run=_make_workflow_run(
run_id="log-run-1",
created_at=datetime(2026, 1, 1, 1, tzinfo=UTC),
finished_at=datetime(2026, 1, 1, 1, 0, 2, tzinfo=UTC),
)
)
# Mock sessionmaker(...).begin() context manager
mock_session = Mock()
mock_db.engine = Mock()
mock_begin = Mock()
mock_begin.__enter__ = Mock(return_value=mock_session)
mock_begin.__exit__ = Mock(return_value=False)
mock_session_factory = Mock()
mock_session_factory.begin.return_value = mock_begin
mock_db.engine = object()
mock_db.session.get.return_value = None
from controllers.service_api.app.workflow import WorkflowAppLogApi
@ -662,8 +821,15 @@ class TestWorkflowAppLogApiGet:
"/workflows/logs?page=1&limit=20",
method="GET",
):
with patch("controllers.service_api.app.workflow.sessionmaker", return_value=mock_session_factory):
with (
patch("controllers.service_api.app.workflow.sessionmaker", _SessionMakerStub),
patch("models.workflow.db", _DbStub()),
patch(
"repositories.factory.DifyAPIRepositoryFactory.create_api_workflow_run_repository",
return_value=mock_repo,
),
):
api = WorkflowAppLogApi()
result = unwrap(api.get)(api, app_model=mock_workflow_app)
result = unwrap(api.get)(api, app_model=workflow_app)
assert result == {"page": 1, "limit": 20, "total": 0, "has_more": False, "data": []}
assert result == _expected_workflow_log_pagination_payload()

43
api/tests/unit_tests/controllers/service_api/app/test_workflow_fields.py
View File

@ -1,25 +1,36 @@
from types import SimpleNamespace
from controllers.service_api.app.workflow import WorkflowRunOutputsField, WorkflowRunStatusField
from controllers.service_api.app.workflow import WorkflowRunResponse
from graphon.enums import WorkflowExecutionStatus
from libs.helper import dump_response
from models.workflow import WorkflowRun
def test_workflow_run_status_field_with_enum() -> None:
field = WorkflowRunStatusField()
obj = SimpleNamespace(status=WorkflowExecutionStatus.PAUSED)
assert field.output("status", obj) == "paused"
def _workflow_run(status: WorkflowExecutionStatus, outputs: str | None = '{"foo": "bar"}') -> WorkflowRun:
return WorkflowRun(
id="run-id",
workflow_id="workflow-id",
status=status,
inputs="{}",
outputs=outputs,
error=None,
total_steps=1,
total_tokens=2,
elapsed_time=3.5,
)
def test_workflow_run_outputs_field_paused_returns_empty() -> None:
field = WorkflowRunOutputsField()
obj = SimpleNamespace(status=WorkflowExecutionStatus.PAUSED, outputs_dict={"foo": "bar"})
def test_workflow_run_serializer_normalizes_status_enum() -> None:
response = dump_response(WorkflowRunResponse, _workflow_run(WorkflowExecutionStatus.PAUSED))
assert field.output("outputs", obj) == {}
assert response["status"] == "paused"
def test_workflow_run_outputs_field_running_returns_outputs() -> None:
field = WorkflowRunOutputsField()
obj = SimpleNamespace(status=WorkflowExecutionStatus.RUNNING, outputs_dict={"foo": "bar"})
def test_workflow_run_serializer_paused_returns_empty_outputs() -> None:
response = dump_response(WorkflowRunResponse, _workflow_run(WorkflowExecutionStatus.PAUSED))
assert field.output("outputs", obj) == {"foo": "bar"}
assert response["outputs"] == {}
def test_workflow_run_serializer_running_returns_outputs() -> None:
response = dump_response(WorkflowRunResponse, _workflow_run(WorkflowExecutionStatus.RUNNING))
assert response["outputs"] == {"foo": "bar"}

23
packages/contracts/generated/api/service/types.gen.ts
View File

@ -20,7 +20,7 @@ export type AgentThought = {
}
export type Annotation = {
content?: string | null
answer?: string | null
created_at?: number | null
hit_count?: number | null
id: string
@ -32,10 +32,15 @@ export type AnnotationCreatePayload = {
question: string
}
export type AnnotationJobStatusResponse = {
error_msg?: string | null
export type AnnotationJobStatusDetailResponse = {
error_msg?: string
job_id: string
job_status: string
job_status: 'completed' | 'error' | 'processing' | 'waiting' | string
}
export type AnnotationJobStatusResponse = {
job_id: string
job_status: 'completed' | 'error' | 'processing' | 'waiting' | string
}
export type AnnotationList = {
@ -1776,7 +1781,7 @@ export type GetAppsAnnotationReplyByActionStatusByJobIdErrors = {
}
export type GetAppsAnnotationReplyByActionStatusByJobIdResponses = {
200: AnnotationJobStatusResponse
200: AnnotationJobStatusDetailResponse
}
export type GetAppsAnnotationReplyByActionStatusByJobIdResponse
@ -1910,7 +1915,9 @@ export type PostChatMessagesErrors = {
}
export type PostChatMessagesResponses = {
200: GeneratedAppResponse
200: {
[key: string]: unknown
}
}
export type PostChatMessagesResponse = PostChatMessagesResponses[keyof PostChatMessagesResponses]
@ -1955,7 +1962,9 @@ export type PostCompletionMessagesErrors = {
}
export type PostCompletionMessagesResponses = {
200: GeneratedAppResponse
200: {
[key: string]: unknown
}
}
export type PostCompletionMessagesResponse

21
packages/contracts/generated/api/service/zod.gen.ts
View File

@ -6,7 +6,7 @@ import * as z from 'zod'
* Annotation
*/
export const zAnnotation = z.object({
content: z.string().nullish(),
answer: z.string().nullish(),
created_at: z.int().nullish(),
hit_count: z.int().nullish(),
id: z.string(),
@ -21,13 +21,21 @@ export const zAnnotationCreatePayload = z.object({
question: z.string(),
})
/**
* AnnotationJobStatusDetailResponse
*/
export const zAnnotationJobStatusDetailResponse = z.object({
error_msg: z.string().optional().default(''),
job_id: z.string(),
job_status: z.union([z.enum(['completed', 'error', 'processing', 'waiting']), z.string()]),
})
/**
* AnnotationJobStatusResponse
*/
export const zAnnotationJobStatusResponse = z.object({
error_msg: z.string().nullish(),
job_id: z.string(),
job_status: z.string(),
job_status: z.union([z.enum(['completed', 'error', 'processing', 'waiting']), z.string()]),
})
/**
@ -2356,7 +2364,8 @@ export const zGetAppsAnnotationReplyByActionStatusByJobIdPath = z.object({
/**
* Successfully retrieved task status.
*/
export const zGetAppsAnnotationReplyByActionStatusByJobIdResponse = zAnnotationJobStatusResponse
export const zGetAppsAnnotationReplyByActionStatusByJobIdResponse
= zAnnotationJobStatusDetailResponse
export const zGetAppsAnnotationsQuery = z.object({
keyword: z.string().optional().default(''),
@ -2414,7 +2423,7 @@ export const zPostChatMessagesBody = zChatRequestPayloadWithUser
* - If `response_mode` is `blocking`, returns `application/json` with a `ChatCompletionResponse` object.
* - If `response_mode` is `streaming`, returns `text/event-stream` with a stream of Server-Sent Events.
*/
export const zPostChatMessagesResponse = zGeneratedAppResponse
export const zPostChatMessagesResponse = z.record(z.string(), z.unknown())
export const zPostChatMessagesByTaskIdStopBody = zRequiredServiceApiUserPayload
@ -2435,7 +2444,7 @@ export const zPostCompletionMessagesBody = zCompletionRequestPayloadWithUser
* - If `response_mode` is `blocking`, returns `application/json` with a `CompletionResponse` object.
* - If `response_mode` is `streaming`, returns `text/event-stream` with a stream of `ChunkCompletionEvent` objects.
*/
export const zPostCompletionMessagesResponse = zGeneratedAppResponse
export const zPostCompletionMessagesResponse = z.record(z.string(), z.unknown())
export const zPostCompletionMessagesByTaskIdStopBody = zRequiredServiceApiUserPayload