314 lines
12 KiB
Python
314 lines
12 KiB
Python
"""Workflow definition / draft / publish API (phase 1 — no execution yet)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from hashlib import sha256
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, HTTPException, Query, Request
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.gateway.deps import get_current_user, get_optional_user_from_request
|
|
from app.gateway.workflow_audit import audit_workflow
|
|
from app.gateway.workflow_resource_checks import build_resource_catalog, make_resource_checker
|
|
from deerflow.config.app_config import get_app_config
|
|
from deerflow.persistence.workflows import (
|
|
WorkflowDraftConflictError,
|
|
WorkflowNotFoundError,
|
|
WorkflowStore,
|
|
)
|
|
from deerflow.workflows.schemas import WorkflowGraph
|
|
from deerflow.workflows.validator import validate_workflow_graph
|
|
|
|
router = APIRouter(prefix="/api/workflows", tags=["workflows"])
|
|
|
|
|
|
class WorkflowCreateRequest(BaseModel):
|
|
name: str = Field(default="未命名工作流", min_length=1, max_length=255)
|
|
description: str = ""
|
|
graph: dict[str, Any] | None = None
|
|
|
|
|
|
class WorkflowPatchRequest(BaseModel):
|
|
name: str | None = Field(default=None, min_length=1, max_length=255)
|
|
description: str | None = None
|
|
|
|
|
|
class DraftPutRequest(BaseModel):
|
|
expected_revision: int = Field(alias="expectedRevision", ge=0)
|
|
graph: dict[str, Any]
|
|
|
|
model_config = {"populate_by_name": True}
|
|
|
|
|
|
class PublishRequest(BaseModel):
|
|
change_note: str = Field(default="", alias="changeNote")
|
|
expected_revision: int | None = Field(default=None, alias="expectedRevision")
|
|
|
|
model_config = {"populate_by_name": True}
|
|
|
|
|
|
class ValidateRequest(BaseModel):
|
|
graph: dict[str, Any] | None = None
|
|
|
|
|
|
def _get_store(request: Request) -> WorkflowStore:
|
|
store = getattr(request.app.state, "workflow_store", None)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="Workflow store not available")
|
|
return store
|
|
|
|
|
|
def _require_enabled() -> None:
|
|
cfg = get_app_config().workflows
|
|
if not cfg.enabled:
|
|
raise HTTPException(status_code=503, detail="Workflow Studio is disabled")
|
|
|
|
|
|
async def _current_user_id(request: Request) -> str:
|
|
return await get_current_user(request)
|
|
|
|
|
|
def _can_access(definition: dict[str, Any], user_id: str, *, write: bool = False) -> bool:
|
|
# Phase 1: owner-only. Shared/org ACLs land later.
|
|
owner = definition.get("owner_id")
|
|
if owner == user_id:
|
|
return True
|
|
# Auth disabled environments often use "default" / anonymous — still owner-scoped.
|
|
if write:
|
|
return False
|
|
return False
|
|
|
|
|
|
async def _load_owned(request: Request, workflow_id: str, *, write: bool = False) -> dict[str, Any]:
|
|
store = _get_store(request)
|
|
user_id = await _current_user_id(request)
|
|
definition = await store.get_definition(workflow_id, include_draft=True)
|
|
if definition is None:
|
|
raise HTTPException(status_code=404, detail="工作流不存在")
|
|
if not _can_access(definition, user_id, write=write):
|
|
# Soft admin bypass for list/read when system_role=admin
|
|
user = await get_optional_user_from_request(request)
|
|
if user is not None and getattr(user, "system_role", None) == "admin" and not write:
|
|
return definition
|
|
if user is not None and getattr(user, "system_role", None) == "admin" and write:
|
|
return definition
|
|
raise HTTPException(status_code=403, detail="无权访问该工作流")
|
|
return definition
|
|
|
|
|
|
def _limits() -> dict[str, int]:
|
|
cfg = get_app_config().workflows
|
|
return {
|
|
"system_max_steps": cfg.max_steps,
|
|
"system_max_loop_iterations": cfg.max_loop_iterations,
|
|
"system_max_parallelism": cfg.max_parallelism,
|
|
}
|
|
|
|
|
|
@router.get("")
|
|
async def list_workflows(
|
|
request: Request,
|
|
include_archived: bool = Query(False, alias="includeArchived"),
|
|
status: str | None = None,
|
|
) -> dict[str, Any]:
|
|
_require_enabled()
|
|
store = _get_store(request)
|
|
user_id = await _current_user_id(request)
|
|
user = await get_optional_user_from_request(request)
|
|
is_admin = user is not None and getattr(user, "system_role", None) == "admin"
|
|
rows = await store.list_definitions(
|
|
owner_id=None if is_admin else user_id,
|
|
status=status,
|
|
include_archived=include_archived,
|
|
)
|
|
if is_admin:
|
|
# Still hide other users' drafts from non-explicit admin listing? Phase 1: admin sees all.
|
|
pass
|
|
else:
|
|
rows = [r for r in rows if r.get("owner_id") == user_id]
|
|
return {"workflows": rows}
|
|
|
|
|
|
@router.post("")
|
|
async def create_workflow(request: Request, body: WorkflowCreateRequest) -> dict[str, Any]:
|
|
_require_enabled()
|
|
store = _get_store(request)
|
|
user_id = await _current_user_id(request)
|
|
created = await store.create_definition(
|
|
{
|
|
"name": body.name,
|
|
"description": body.description,
|
|
"owner_id": user_id,
|
|
"draft_graph": body.graph,
|
|
}
|
|
)
|
|
audit_workflow("workflow.create", user_id=user_id, workflowId=created["id"])
|
|
return created
|
|
|
|
|
|
@router.get("/{workflow_id}")
|
|
async def get_workflow(request: Request, workflow_id: str) -> dict[str, Any]:
|
|
_require_enabled()
|
|
return await _load_owned(request, workflow_id, write=False)
|
|
|
|
|
|
@router.patch("/{workflow_id}")
|
|
async def patch_workflow(request: Request, workflow_id: str, body: WorkflowPatchRequest) -> dict[str, Any]:
|
|
_require_enabled()
|
|
await _load_owned(request, workflow_id, write=True)
|
|
store = _get_store(request)
|
|
updated = await store.update_definition(
|
|
workflow_id,
|
|
body.model_dump(exclude_unset=True),
|
|
)
|
|
if updated is None:
|
|
raise HTTPException(status_code=404, detail="工作流不存在")
|
|
return updated
|
|
|
|
|
|
@router.delete("/{workflow_id}")
|
|
async def archive_workflow(request: Request, workflow_id: str) -> dict[str, Any]:
|
|
_require_enabled()
|
|
await _load_owned(request, workflow_id, write=True)
|
|
store = _get_store(request)
|
|
archived = await store.archive_definition(workflow_id)
|
|
if archived is None:
|
|
raise HTTPException(status_code=404, detail="工作流不存在")
|
|
audit_workflow("workflow.archive", user_id=await _current_user_id(request), workflowId=workflow_id)
|
|
return archived
|
|
|
|
|
|
@router.post("/{workflow_id}/copy")
|
|
async def copy_workflow(request: Request, workflow_id: str) -> dict[str, Any]:
|
|
_require_enabled()
|
|
definition = await _load_owned(request, workflow_id, write=False)
|
|
store = _get_store(request)
|
|
user_id = await _current_user_id(request)
|
|
created = await store.create_definition(
|
|
{
|
|
"name": f"{definition.get('name') or '未命名工作流'} 副本",
|
|
"description": definition.get("description") or "",
|
|
"owner_id": user_id,
|
|
"draft_graph": definition.get("draft_graph"),
|
|
}
|
|
)
|
|
audit_workflow("workflow.copy", user_id=user_id, workflowId=created["id"], sourceWorkflowId=workflow_id)
|
|
return created
|
|
|
|
|
|
@router.put("/{workflow_id}/draft")
|
|
async def put_draft(request: Request, workflow_id: str, body: DraftPutRequest) -> dict[str, Any]:
|
|
_require_enabled()
|
|
await _load_owned(request, workflow_id, write=True)
|
|
store = _get_store(request)
|
|
user_id = await _current_user_id(request)
|
|
try:
|
|
# Light structural parse so obviously broken drafts fail early; full publish
|
|
# checks run on /validate and /publish.
|
|
WorkflowGraph.model_validate(body.graph)
|
|
except Exception as exc: # noqa: BLE001 — surface pydantic as 400
|
|
raise HTTPException(status_code=400, detail=f"图结构无效: {exc}") from exc
|
|
try:
|
|
return await store.save_draft(
|
|
workflow_id,
|
|
expected_revision=body.expected_revision,
|
|
graph=body.graph,
|
|
updated_by=user_id,
|
|
)
|
|
except WorkflowDraftConflictError as exc:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail={
|
|
"code": "WORKFLOW_DRAFT_CONFLICT",
|
|
"message": "草稿版本冲突，请刷新后重试或另存副本",
|
|
"currentRevision": exc.current_revision,
|
|
},
|
|
) from exc
|
|
except WorkflowNotFoundError as exc:
|
|
raise HTTPException(status_code=404, detail="工作流不存在") from exc
|
|
|
|
|
|
@router.post("/{workflow_id}/validate")
|
|
async def validate_workflow(request: Request, workflow_id: str, body: ValidateRequest | None = None) -> dict[str, Any]:
|
|
_require_enabled()
|
|
definition = await _load_owned(request, workflow_id, write=False)
|
|
graph = (body.graph if body and body.graph is not None else definition.get("draft_graph")) or {}
|
|
issues = validate_workflow_graph(graph, **_limits())
|
|
return {
|
|
"ok": len(issues) == 0,
|
|
"issues": [i.model_dump(by_alias=True) for i in issues],
|
|
"graphHash": WorkflowGraph.model_validate(graph).graph_hash() if not issues else None,
|
|
}
|
|
|
|
|
|
@router.post("/{workflow_id}/publish")
|
|
async def publish_workflow(request: Request, workflow_id: str, body: PublishRequest | None = None) -> dict[str, Any]:
|
|
_require_enabled()
|
|
definition = await _load_owned(request, workflow_id, write=True)
|
|
body = body or PublishRequest()
|
|
if body.expected_revision is not None and body.expected_revision != definition.get("draft_revision"):
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail={
|
|
"code": "WORKFLOW_DRAFT_CONFLICT",
|
|
"message": "发布前草稿版本已变化",
|
|
"currentRevision": definition.get("draft_revision"),
|
|
},
|
|
)
|
|
graph_raw = definition.get("draft_graph") or {}
|
|
# Same resource/permission checks as the studio facade (§5.3): the standard
|
|
# publish path must not accept graphs referencing invisible agents or
|
|
# forbidden resources.
|
|
catalog = await build_resource_catalog(request, user_id=await _current_user_id(request))
|
|
checker = make_resource_checker(catalog, config=get_app_config().workflows, current_workflow_id=workflow_id)
|
|
issues = validate_workflow_graph(graph_raw, resource_checker=checker, **_limits())
|
|
if issues:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail={
|
|
"code": "WORKFLOW_SCHEMA_INVALID",
|
|
"message": "发布校验未通过",
|
|
"issues": [i.model_dump(by_alias=True) for i in issues],
|
|
},
|
|
)
|
|
graph = WorkflowGraph.model_validate(graph_raw)
|
|
store = _get_store(request)
|
|
user_id = await _current_user_id(request)
|
|
canvas_text = str(definition.get("draft_canvas_schema") or "{}")
|
|
version = await store.publish_version(
|
|
workflow_id,
|
|
graph=graph.model_dump(by_alias=True),
|
|
graph_hash=graph.graph_hash(),
|
|
published_by=user_id,
|
|
change_note=body.change_note,
|
|
canvas_schema_hash=sha256(canvas_text.encode("utf-8")).hexdigest(),
|
|
)
|
|
audit_workflow(
|
|
"workflow.publish",
|
|
user_id=user_id,
|
|
workflowId=workflow_id,
|
|
versionId=version.get("id"),
|
|
graphHash=graph.graph_hash(),
|
|
)
|
|
return version
|
|
|
|
|
|
@router.get("/{workflow_id}/versions")
|
|
async def list_versions(request: Request, workflow_id: str) -> dict[str, Any]:
|
|
_require_enabled()
|
|
await _load_owned(request, workflow_id, write=False)
|
|
store = _get_store(request)
|
|
return {"versions": await store.list_versions(workflow_id)}
|
|
|
|
|
|
@router.get("/{workflow_id}/versions/{version_id}")
|
|
async def get_version(request: Request, workflow_id: str, version_id: str) -> dict[str, Any]:
|
|
_require_enabled()
|
|
await _load_owned(request, workflow_id, write=False)
|
|
store = _get_store(request)
|
|
version = await store.get_version(version_id)
|
|
if version is None or version.get("workflow_id") != workflow_id:
|
|
raise HTTPException(status_code=404, detail="版本不存在")
|
|
return version
|