diff --git a/backend/packages/app/src/windup_app/bootstrap/app.py b/backend/packages/app/src/windup_app/bootstrap/app.py index 50ce41a8..fba1ea8d 100644 --- a/backend/packages/app/src/windup_app/bootstrap/app.py +++ b/backend/packages/app/src/windup_app/bootstrap/app.py @@ -18,11 +18,13 @@ from windup_app.server.character.model import Character # noqa: F401 from windup_app.server.project.model import Project # noqa: F401 from windup_app.server.user.model import User # noqa: F401 +from windup_app.server.workflow_run.model import WorkflowRun # noqa: F401 from windup_app.web.api.auth import router as auth_router from windup_app.web.api.character import router as character_router from windup_app.web.api.generation import router as generation_router from windup_app.web.api.media import router as media_router from windup_app.web.api.project import router as project_router +from windup_app.web.api.workflow_run import router as workflow_run_router from windup_app.web.handler.exception_handlers import register_exception_handlers from windup_app.web.middleware.auth import AuthMiddleware from windup_app.web.middleware.ratelimit import RateLimitMiddleware @@ -85,6 +87,7 @@ def create_app() -> FastAPI: app.include_router(auth_router) app.include_router(project_router) app.include_router(character_router) + app.include_router(workflow_run_router) app.include_router(media_router) app.include_router(generation_router) register_exception_handlers(app) diff --git a/backend/packages/app/src/windup_app/server/workflow_run/interface.py b/backend/packages/app/src/windup_app/server/workflow_run/interface.py index c4d65e3b..3fe597a1 100644 --- a/backend/packages/app/src/windup_app/server/workflow_run/interface.py +++ b/backend/packages/app/src/windup_app/server/workflow_run/interface.py @@ -18,20 +18,18 @@ from abc import ABC, abstractmethod -from windup_app.server.workflow_run.model import ( - RunStatus, - WorkflowRun, -) +from sqlalchemy.orm import Session + +from windup_app.server.workflow_run.model import RunStatus, WorkflowRun class WorkflowRunService(ABC): """执行记录用例的抽象边界。""" - # -- 执行记录 CRUD -------------------------------------------------------- - @abstractmethod def create_run( self, + session: Session, *, project_id: int, nodes: list | None = None, @@ -42,22 +40,35 @@ def create_run( """ @abstractmethod - def get_run(self, run_id: int) -> WorkflowRun | None: + def get_run(self, session: Session, run_id: int) -> WorkflowRun | None: """获取执行记录详情（含 nodes JSONB）。""" + @abstractmethod + def list_runs( + self, + session: Session, + *, + project_id: int, + page: int = 1, + page_size: int = 20, + ) -> tuple[list[WorkflowRun], int]: + """分页查询项目下的执行记录，返回 (当前页数据, 总数)。""" + @abstractmethod def update_run( self, + session: Session, run_id: int, *, nodes: list | None = None, status: RunStatus | None = None, - ) -> WorkflowRun: - """全量更新执行记录。 + ) -> WorkflowRun | None: + """更新执行记录。 前端维护节点树后，通过此接口全量写回。 + 返回更新后的记录；不存在时返回 None。 """ @abstractmethod - def delete_run(self, run_id: int) -> None: - """软删除执行记录。""" + def delete_run(self, session: Session, run_id: int) -> bool: + """软删除执行记录。返回是否找到。""" diff --git a/backend/packages/app/src/windup_app/server/workflow_run/model.py b/backend/packages/app/src/windup_app/server/workflow_run/model.py index e3ff7226..d9156b1f 100644 --- a/backend/packages/app/src/windup_app/server/workflow_run/model.py +++ b/backend/packages/app/src/windup_app/server/workflow_run/model.py @@ -6,10 +6,15 @@ from __future__ import annotations -from dataclasses import dataclass, field from datetime import datetime, timezone from enum import StrEnum +from sqlalchemy import BigInteger, DateTime, Integer, JSON, String +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from windup_framework.db import Base + # -- 枚举 ---------------------------------------------------------------- @@ -21,19 +26,42 @@ class RunStatus(StrEnum): SOFT_DELETED = "soft_deleted" -# -- 执行记录 ------------------------------------------------------------- +# -- ORM ----------------------------------------------------------------- -@dataclass -class WorkflowRun: - """执行记录——前端维护的节点树的持久化容器。 +class WorkflowRun(Base): + """执行记录表——前端维护的节点树的持久化容器。 后端不校验 nodes 内部结构，仅做全量读写。 """ - id: int | None = None - project_id: int = 0 - nodes: list = field(default_factory=list) # 节点树（前端自定义结构，后端不校验） - status: RunStatus = RunStatus.ACTIVE - version: int = 1 - created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + __tablename__ = "windup_workflow_run" + + id: Mapped[int] = mapped_column( + BigInteger().with_variant(Integer, "sqlite"), + primary_key=True, + autoincrement=True, + ) + + project_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + + # 节点树（前端自定义结构，后端不校验）；Postgres 上 JSONB，SQLite 上 JSON。 + nodes: Mapped[list] = mapped_column( + JSON().with_variant(JSONB, "postgresql"), + nullable=False, + default=list, + ) + + status: Mapped[str] = mapped_column( + String(20), nullable=False, default=RunStatus.ACTIVE.value, + ) + + version: Mapped[int] = mapped_column( + Integer, nullable=False, default=1, + ) + + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + nullable=False, + default=lambda: datetime.now(timezone.utc), + ) diff --git a/backend/packages/app/src/windup_app/server/workflow_run/service.py b/backend/packages/app/src/windup_app/server/workflow_run/service.py new file mode 100644 index 00000000..916e8b64 --- /dev/null +++ b/backend/packages/app/src/windup_app/server/workflow_run/service.py @@ -0,0 +1,98 @@ +"""工作流执行记录领域服务的 SQLAlchemy 实现。 + +:class:`SqlAlchemyWorkflowRunService` 继承 :class:`WorkflowRunService` 接口，用同步 +SQLAlchemy session 落库。无状态：``session`` 由调用方按请求传入，本对象可作 +模块级单例（:data:`service`）。 + +事务边界由 ``windup_framework.db.get_session`` 依赖负责——成功 commit、异常 +rollback，故本实现只 ``flush``（把变更发到当前事务、取回生成的主键），不 commit。 +""" + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from windup_app.server.workflow_run.interface import WorkflowRunService +from windup_app.server.workflow_run.model import RunStatus, WorkflowRun + + +class SqlAlchemyWorkflowRunService(WorkflowRunService): + """基于 SQLAlchemy session 的执行记录 CRUD 实现。""" + + def create_run( + self, + session: Session, + *, + project_id: int, + nodes: list | None = None, + ) -> WorkflowRun: + run = WorkflowRun( + project_id=project_id, + nodes=nodes or [], + ) + session.add(run) + session.flush() + return run + + def get_run(self, session: Session, run_id: int) -> WorkflowRun | None: + return session.get(WorkflowRun, run_id) + + def list_runs( + self, + session: Session, + *, + project_id: int, + page: int = 1, + page_size: int = 20, + ) -> tuple[list[WorkflowRun], int]: + """分页查询项目下的执行记录，返回 (当前页数据, 总数)。""" + count_stmt = ( + select(func.count()) + .select_from(WorkflowRun) + .where( + WorkflowRun.project_id == project_id, + WorkflowRun.status != RunStatus.SOFT_DELETED.value, + ) + ) + stmt = ( + select(WorkflowRun) + .where( + WorkflowRun.project_id == project_id, + WorkflowRun.status != RunStatus.SOFT_DELETED.value, + ) + .order_by(WorkflowRun.id.desc()) + .offset((page - 1) * page_size) + .limit(page_size) + ) + total = session.scalar(count_stmt) or 0 + items = list(session.scalars(stmt)) + return items, total + + def update_run( + self, + session: Session, + run_id: int, + *, + nodes: list | None = None, + status: RunStatus | None = None, + ) -> WorkflowRun | None: + run = session.get(WorkflowRun, run_id) + if run is None: + return None + if nodes is not None: + run.nodes = nodes + if status is not None: + run.status = status.value + run.version += 1 + session.flush() + return run + + def delete_run(self, session: Session, run_id: int) -> bool: + run = session.get(WorkflowRun, run_id) + if run is None: + return False + run.status = RunStatus.SOFT_DELETED.value + session.flush() + return True + + +service = SqlAlchemyWorkflowRunService() diff --git a/backend/packages/app/src/windup_app/web/api/workflow_run.py b/backend/packages/app/src/windup_app/web/api/workflow_run.py index 0cfc61a0..23a2da71 100644 --- a/backend/packages/app/src/windup_app/web/api/workflow_run.py +++ b/backend/packages/app/src/windup_app/web/api/workflow_run.py @@ -1,14 +1,12 @@ """工作流执行记录 API。 -契约层：定义端点和请求/响应模型，与 server 层解耦。 -实际逻辑由 server 层实现，本文件只做参数校验和格式转换。 - 端点一览 -------- POST /workflow-runs 创建执行记录 -GET /workflow-runs/{id} 获取执行记录（含 nodes） -PATCH /workflow-runs/{id} 全量更新（含 nodes） -DELETE /workflow-runs/{id} 软删除 +GET /workflow-runs?project_id=... 分页列表 +GET /workflow-runs/{id} 获取执行记录（含 nodes） +PATCH /workflow-runs/{id} 全量更新（含 nodes） +DELETE /workflow-runs/{id} 软删除 设计原则 -------- @@ -20,69 +18,171 @@ import logging -from fastapi import APIRouter, Depends +from fastapi import APIRouter, Depends, Query, Request +from pydantic import BaseModel, ConfigDict, Field from sqlalchemy.orm import Session -from windup_common.result import Response +from windup_common.enums.biz_code import BizCode +from windup_common.exceptions import BizException +from windup_common.result import ListResponse, Response from windup_framework.db import get_session -from windup_app.server.workflow_run.schema import ( - WorkflowRunCreateRequest, - WorkflowRunOut, - WorkflowRunUpdateRequest, -) +from windup_app.server.project.model import Project +from windup_app.server.workflow_run.model import RunStatus +from windup_app.server.workflow_run.service import service logger = logging.getLogger("windup.workflow_run.api") router = APIRouter(prefix="/workflow-runs", tags=["workflow-run"]) -# ── 执行记录 CRUD ─────────────────────────────────────────────────────────── +# ── 请求 / 响应模型 ───────────────────────────────────────────────────────── + + +class WorkflowRunCreate(BaseModel): + """创建执行记录。""" + + project_id: int = Field(gt=0) + nodes: list = Field( + default_factory=list, + description="节点树（前端自定义结构，后端不校验）", + ) + + +class WorkflowRunUpdate(BaseModel): + """全量更新执行记录。""" + + nodes: list | None = Field( + default=None, + description="节点树（前端自定义结构，后端不校验）", + ) + status: str | None = Field( + default=None, + description="状态：active / soft_deleted", + ) + + +class WorkflowRunOut(BaseModel): + """执行记录响应。""" + + model_config = ConfigDict(from_attributes=True) + + id: int + project_id: int + nodes: list = Field(default_factory=list, description="节点树") + status: str + version: int + + +# ── 归属校验 ───────────────────────────────────────────────────────────────── + + +def _get_project_or_raise( + session: Session, project_id: int, user_id: int, +) -> Project: + """校验项目存在且属于当前用户。""" + project = session.get(Project, project_id) + if project is None or project.user_id != user_id: + raise BizException("项目不存在", code=BizCode.NOT_FOUND) + return project + + +def _get_run_with_auth( + session: Session, run_id: int, user_id: int, +): + """获取执行记录并校验其所属项目属于当前用户。""" + run = service.get_run(session, run_id) + if run is None: + raise BizException("执行记录不存在", code=BizCode.NOT_FOUND) + project = session.get(Project, run.project_id) + if project is None or project.user_id != user_id: + raise BizException("执行记录不存在", code=BizCode.NOT_FOUND) + return run + + +# ── 端点 ───────────────────────────────────────────────────────────────────── @router.post("", response_model=Response[WorkflowRunOut]) def create_run( - body: WorkflowRunCreateRequest, + body: WorkflowRunCreate, + request: Request, session: Session = Depends(get_session), ) -> Response[WorkflowRunOut]: - """创建执行记录。 - - nodes 为前端定义的初始节点树（可选）。 - """ - # TODO: service.create_run - raise NotImplementedError + """创建执行记录。""" + user_id = request.state.current_user.id + _get_project_or_raise(session, body.project_id, user_id) + run = service.create_run(session, project_id=body.project_id, nodes=body.nodes) + return Response.success(WorkflowRunOut.model_validate(run), message="创建成功") + + +@router.get("", response_model=ListResponse[WorkflowRunOut]) +def list_runs( + project_id: int = Query(..., gt=0), + request: Request = None, + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + session: Session = Depends(get_session), +) -> ListResponse[WorkflowRunOut]: + """分页查询项目下的执行记录。""" + user_id = request.state.current_user.id + _get_project_or_raise(session, project_id, user_id) + items, total = service.list_runs( + session, project_id=project_id, page=page, page_size=page_size, + ) + return ListResponse.success( + [WorkflowRunOut.model_validate(r) for r in items], + total=total, + page=page, + page_size=page_size, + ) @router.get("/{run_id}", response_model=Response[WorkflowRunOut]) def get_run( run_id: int, + request: Request, session: Session = Depends(get_session), ) -> Response[WorkflowRunOut]: """获取执行记录详情（含 nodes JSONB）。""" - # TODO: service.get_run - raise NotImplementedError + user_id = request.state.current_user.id + run = _get_run_with_auth(session, run_id, user_id) + return Response.success(WorkflowRunOut.model_validate(run)) @router.patch("/{run_id}", response_model=Response[WorkflowRunOut]) def update_run( run_id: int, - body: WorkflowRunUpdateRequest, + body: WorkflowRunUpdate, + request: Request, session: Session = Depends(get_session), ) -> Response[WorkflowRunOut]: - """全量更新执行记录。 + """全量更新执行记录。""" + user_id = request.state.current_user.id + _get_run_with_auth(session, run_id, user_id) + + status = None + if body.status is not None: + try: + status = RunStatus(body.status) + except ValueError: + raise BizException( + f"无效状态: {body.status}，可选: active / soft_deleted", + code=BizCode.BAD_REQUEST, + ) from None - 前端维护节点树后，通过此接口全量写回。 - 后端不校验 nodes 内部结构。 - """ - # TODO: service.update_run - raise NotImplementedError + run = service.update_run(session, run_id, nodes=body.nodes, status=status) + return Response.success(WorkflowRunOut.model_validate(run), message="更新成功") @router.delete("/{run_id}", response_model=Response[None]) def delete_run( run_id: int, + request: Request, session: Session = Depends(get_session), ) -> Response[None]: """软删除执行记录。""" - # TODO: service.delete_run - raise NotImplementedError + user_id = request.state.current_user.id + _get_run_with_auth(session, run_id, user_id) + service.delete_run(session, run_id) + return Response.success(None, message="删除成功") diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index a3d471e8..6a120038 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -15,6 +15,7 @@ from windup_app.server.character.model import Character from windup_app.server.project.model import Project from windup_app.server.user.model import User +from windup_app.server.workflow_run.model import WorkflowRun from windup_app.server.user.service import create_access_token from windup_framework.db import Base, get_session @@ -32,7 +33,7 @@ def _make_engine(): def engine(): """建好 ``windup_project`` 和 ``windup_user`` 表的内存 engine。""" engine = _make_engine() - Base.metadata.create_all(engine, tables=[Project.__table__, User.__table__, Character.__table__]) + Base.metadata.create_all(engine, tables=[Project.__table__, User.__table__, Character.__table__, WorkflowRun.__table__]) yield engine engine.dispose() diff --git a/backend/tests/test_workflow_run_api.py b/backend/tests/test_workflow_run_api.py new file mode 100644 index 00000000..6546e3a8 --- /dev/null +++ b/backend/tests/test_workflow_run_api.py @@ -0,0 +1,245 @@ +"""工作流执行记录 CRUD API 集成测试。""" + + +def _create_project(auth_client, name: str = "默认项目") -> dict: + """创建一个项目并返回响应 data。""" + return auth_client.post("/projects", json={ + "project_name": name, + "character_perspective": 1, + "directional_movement": 2, + "sprite_width": 64, + "sprite_height": 64, + }).json()["data"] + + +def _payload(project_id: int, **overrides): + """构造合法的创建执行记录请求体。""" + base = { + "project_id": project_id, + } + base.update(overrides) + return base + + +# -- POST /workflow-runs ------------------------------------------------------ + + +def test_create_success(auth_client): + project = _create_project(auth_client) + resp = auth_client.post("/workflow-runs", json=_payload(project["id"])) + + assert resp.status_code == 200 + body = resp.json() + assert body["code"] == 200 + assert body["data"]["project_id"] == project["id"] + assert body["data"]["nodes"] == [] + assert body["data"]["status"] == "active" + assert body["data"]["version"] == 1 + + +def test_create_with_nodes(auth_client): + project = _create_project(auth_client) + nodes = [{"id": "n1", "type": "start"}, {"id": "n2", "type": "end"}] + resp = auth_client.post( + "/workflow-runs", json=_payload(project["id"], nodes=nodes), + ) + + assert resp.json()["code"] == 200 + assert resp.json()["data"]["nodes"] == nodes + + +def test_create_under_other_users_project_returns_404(auth_client, auth_client_b): + """用户 B 不能在用户 A 的项目下创建执行记录。""" + project = _create_project(auth_client) + resp = auth_client_b.post("/workflow-runs", json=_payload(project["id"])) + + assert resp.json()["code"] == 404 + assert resp.json()["message"] == "项目不存在" + + +# -- GET /workflow-runs -------------------------------------------------------- + + +def test_list_empty(auth_client): + project = _create_project(auth_client) + resp = auth_client.get( + "/workflow-runs", params={"project_id": project["id"]}, + ) + + body = resp.json() + assert body["code"] == 200 + assert body["data"] == [] + assert body["total"] == 0 + + +def test_list_paginates(auth_client): + project = _create_project(auth_client) + for _ in range(3): + auth_client.post("/workflow-runs", json=_payload(project["id"])) + + resp = auth_client.get( + "/workflow-runs", + params={"project_id": project["id"], "page": 1, "page_size": 2}, + ) + + body = resp.json() + assert body["total"] == 3 + assert len(body["data"]) == 2 + + +def test_list_excludes_soft_deleted(auth_client): + project = _create_project(auth_client) + r1 = auth_client.post("/workflow-runs", json=_payload(project["id"])).json()["data"] + auth_client.post("/workflow-runs", json=_payload(project["id"])) + + # 软删除 r1 + auth_client.delete(f"/workflow-runs/{r1['id']}") + + resp = auth_client.get( + "/workflow-runs", params={"project_id": project["id"]}, + ) + assert resp.json()["total"] == 1 + + +def test_list_other_users_project_returns_404(auth_client, auth_client_b): + """用户 B 不能列出用户 A 项目的执行记录。""" + project = _create_project(auth_client) + auth_client.post("/workflow-runs", json=_payload(project["id"])) + + resp = auth_client_b.get( + "/workflow-runs", params={"project_id": project["id"]}, + ) + assert resp.json()["code"] == 404 + + +# -- GET /workflow-runs/{id} --------------------------------------------------- + + +def test_get_success(auth_client): + project = _create_project(auth_client) + created = auth_client.post( + "/workflow-runs", json=_payload(project["id"]), + ).json()["data"] + + resp = auth_client.get(f"/workflow-runs/{created['id']}") + + assert resp.json()["code"] == 200 + assert resp.json()["data"]["id"] == created["id"] + + +def test_get_not_found_returns_404(auth_client): + resp = auth_client.get("/workflow-runs/99999") + + assert resp.json()["code"] == 404 + assert resp.json()["message"] == "执行记录不存在" + + +def test_get_other_users_run_returns_404(auth_client, auth_client_b): + """用户 B 不能查看用户 A 的执行记录。""" + project = _create_project(auth_client) + created = auth_client.post( + "/workflow-runs", json=_payload(project["id"]), + ).json()["data"] + + resp = auth_client_b.get(f"/workflow-runs/{created['id']}") + + assert resp.json()["code"] == 404 + + +# -- PATCH /workflow-runs/{id} ------------------------------------------------- + + +def test_update_nodes(auth_client): + project = _create_project(auth_client) + created = auth_client.post( + "/workflow-runs", json=_payload(project["id"]), + ).json()["data"] + + new_nodes = [{"id": "n1", "type": "action"}] + resp = auth_client.patch( + f"/workflow-runs/{created['id']}", json={"nodes": new_nodes}, + ) + + assert resp.json()["code"] == 200 + assert resp.json()["data"]["nodes"] == new_nodes + assert resp.json()["data"]["version"] == 2 + + +def test_update_status(auth_client): + project = _create_project(auth_client) + created = auth_client.post( + "/workflow-runs", json=_payload(project["id"]), + ).json()["data"] + + resp = auth_client.patch( + f"/workflow-runs/{created['id']}", json={"status": "soft_deleted"}, + ) + + assert resp.json()["code"] == 200 + assert resp.json()["data"]["status"] == "soft_deleted" + + +def test_update_invalid_status_returns_400(auth_client): + project = _create_project(auth_client) + created = auth_client.post( + "/workflow-runs", json=_payload(project["id"]), + ).json()["data"] + + resp = auth_client.patch( + f"/workflow-runs/{created['id']}", json={"status": "bogus"}, + ) + + assert resp.json()["code"] == 400 + + +def test_update_other_users_run_returns_404(auth_client, auth_client_b): + """用户 B 不能修改用户 A 的执行记录。""" + project = _create_project(auth_client) + created = auth_client.post( + "/workflow-runs", json=_payload(project["id"]), + ).json()["data"] + + resp = auth_client_b.patch( + f"/workflow-runs/{created['id']}", json={"nodes": []}, + ) + + assert resp.json()["code"] == 404 + + +# -- DELETE /workflow-runs/{id} ------------------------------------------------ + + +def test_delete_success(auth_client): + project = _create_project(auth_client) + created = auth_client.post( + "/workflow-runs", json=_payload(project["id"]), + ).json()["data"] + + resp = auth_client.delete(f"/workflow-runs/{created['id']}") + + assert resp.json()["code"] == 200 + assert resp.json()["message"] == "删除成功" + + # 删除后列表不包含该记录 + resp = auth_client.get( + "/workflow-runs", params={"project_id": project["id"]}, + ) + assert resp.json()["total"] == 0 + + +def test_delete_not_found_returns_404(auth_client): + resp = auth_client.delete("/workflow-runs/99999") + + assert resp.json()["code"] == 404 + + +def test_delete_other_users_run_returns_404(auth_client, auth_client_b): + """用户 B 不能删除用户 A 的执行记录。""" + project = _create_project(auth_client) + created = auth_client.post( + "/workflow-runs", json=_payload(project["id"]), + ).json()["data"] + + resp = auth_client_b.delete(f"/workflow-runs/{created['id']}") + + assert resp.json()["code"] == 404