mirror of
https://github.com/langgenius/dify.git
synced 2026-05-20 08:46:57 +08:00
cf5ebe9430
Ports service_api/app/{completion,workflow}.py to bearer-authed
/openapi/v1/apps/<app_id>/{info,chat-messages,completion-messages,workflows/run}.
Architecture:
- New controllers/openapi/auth/ package: Pipeline + Step protocol over
one mutable Context. Endpoints attach via @APP_PIPELINE.guard(scope=...)
— single attachment point; forgetting auth is structurally impossible.
- Pipeline order: BearerCheck -> ScopeCheck -> AppResolver -> AppAuthzCheck
-> CallerMount.
- Strategies vary along independent axes: AclStrategy (EE webapp-auth inner
API) vs MembershipStrategy (CE TenantAccountJoin); AccountMounter vs
EndUserMounter dispatched by SubjectType.
- App is in URL path (not header). Each non-GET has typed Pydantic Request;
each non-SSE response has typed Pydantic Response. Bearer-as-identity:
body 'user' field stripped, ignored if present.
Adds InvokeFrom.OPENAPI enum variant. Emits app.run.openapi audit log
on successful invocation via standard logger extra={"audit": True, ...}
convention.
133 lines
4.3 KiB
Python
133 lines
4.3 KiB
Python
"""POST /openapi/v1/apps/<app_id>/workflows/run — port of
|
|
service_api/app/workflow.py:WorkflowRunApi."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any, Literal
|
|
|
|
from flask import request
|
|
from flask_restx import Resource
|
|
from pydantic import BaseModel, ValidationError
|
|
from werkzeug.exceptions import BadRequest, InternalServerError, NotFound
|
|
|
|
from controllers.common.controller_schemas import WorkflowRunPayload as WorkflowRunPayloadBase
|
|
from controllers.openapi import openapi_ns
|
|
from controllers.openapi._audit import emit_app_run
|
|
from controllers.openapi.auth.composition import APP_PIPELINE
|
|
from controllers.service_api.app.error import (
|
|
CompletionRequestError,
|
|
NotWorkflowAppError,
|
|
ProviderModelCurrentlyNotSupportError,
|
|
ProviderNotInitializeError,
|
|
ProviderQuotaExceededError,
|
|
)
|
|
from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
|
|
from core.app.entities.app_invoke_entities import InvokeFrom
|
|
from core.errors.error import (
|
|
ModelCurrentlyNotSupportError,
|
|
ProviderTokenNotInitError,
|
|
QuotaExceededError,
|
|
)
|
|
from graphon.model_runtime.errors.invoke import InvokeError
|
|
from libs import helper
|
|
from models.model import App, AppMode
|
|
from services.app_generate_service import AppGenerateService
|
|
from services.errors.app import (
|
|
IsDraftWorkflowError,
|
|
WorkflowIdFormatError,
|
|
WorkflowNotFoundError,
|
|
)
|
|
from services.errors.llm import InvokeRateLimitError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class WorkflowRunRequest(WorkflowRunPayloadBase):
|
|
response_mode: Literal["blocking", "streaming"] | None = None
|
|
|
|
|
|
class WorkflowRunData(BaseModel):
|
|
id: str
|
|
workflow_id: str
|
|
status: str
|
|
outputs: dict[str, Any] = {}
|
|
error: str | None = None
|
|
elapsed_time: float | None = None
|
|
total_tokens: int | None = None
|
|
total_steps: int | None = None
|
|
created_at: int | None = None
|
|
finished_at: int | None = None
|
|
|
|
|
|
class WorkflowRunResponse(BaseModel):
|
|
workflow_run_id: str
|
|
task_id: str
|
|
data: WorkflowRunData
|
|
|
|
|
|
def _unpack_app(app_model):
|
|
return app_model
|
|
|
|
|
|
def _unpack_caller(caller):
|
|
return caller
|
|
|
|
|
|
@openapi_ns.route("/apps/<string:app_id>/workflows/run")
|
|
class WorkflowRunApi(Resource):
|
|
@APP_PIPELINE.guard(scope="apps:run")
|
|
def post(self, app_id: str, app_model: App, caller, caller_kind: str):
|
|
app = _unpack_app(app_model)
|
|
if AppMode.value_of(app.mode) != AppMode.WORKFLOW:
|
|
raise NotWorkflowAppError()
|
|
|
|
body = request.get_json(silent=True) or {}
|
|
body.pop("user", None)
|
|
try:
|
|
payload = WorkflowRunRequest.model_validate(body)
|
|
except ValidationError as exc:
|
|
raise BadRequest(str(exc))
|
|
args = payload.model_dump(exclude_none=True)
|
|
streaming = payload.response_mode == "streaming"
|
|
|
|
try:
|
|
response = AppGenerateService.generate(
|
|
app_model=app,
|
|
user=_unpack_caller(caller),
|
|
args=args,
|
|
invoke_from=InvokeFrom.OPENAPI,
|
|
streaming=streaming,
|
|
)
|
|
except WorkflowNotFoundError as ex:
|
|
raise NotFound(str(ex))
|
|
except (IsDraftWorkflowError, WorkflowIdFormatError) as ex:
|
|
raise BadRequest(str(ex))
|
|
except ProviderTokenNotInitError as ex:
|
|
raise ProviderNotInitializeError(ex.description)
|
|
except QuotaExceededError:
|
|
raise ProviderQuotaExceededError()
|
|
except ModelCurrentlyNotSupportError:
|
|
raise ProviderModelCurrentlyNotSupportError()
|
|
except InvokeRateLimitError as ex:
|
|
raise InvokeRateLimitHttpError(ex.description)
|
|
except InvokeError as e:
|
|
raise CompletionRequestError(e.description)
|
|
except ValueError:
|
|
raise
|
|
except Exception:
|
|
logger.exception("internal server error.")
|
|
raise InternalServerError()
|
|
|
|
emit_app_run(
|
|
app_id=app.id,
|
|
tenant_id=app.tenant_id,
|
|
caller_kind=caller_kind,
|
|
mode=str(app.mode),
|
|
)
|
|
|
|
if streaming:
|
|
return helper.compact_generate_response(response)
|
|
|
|
body_dict = response[0] if isinstance(response, tuple) else response
|
|
return WorkflowRunResponse.model_validate(body_dict).model_dump(mode="json"), 200
|