M4:单个工作流接口的 run_count 和列表接口按同一个函数算

验收时 GET /api/workflows/{id} 的 run_count 恒为 0,同一时刻列表接口里同一个工作流是 2 次、4 次:只有列表接口按
Run.workflow_id 计数,详情、保存(PATCH)、恢复旧版、新建、复制直接返回数据库对象,落在 WorkflowOut 的默认值 0 上。
计数抽成 _run_counts,列表和返回单个工作流的接口(_workflow_out)都走它,次数一致;新建、复制出来的工作流照实是 0。
This commit is contained in:
ThearchyHelios
2026-10-06 23:56:31 +08:00
parent 58a6759fe2
commit e410d2a6a7
2 changed files with 111 additions and 13 deletions
+27 -13
View File
@@ -65,14 +65,28 @@ class WorkflowOut(BaseModel):
model_config = {"from_attributes": True}
async def _run_counts(session: AsyncSession, ids: list[str] | None = None) -> dict[str, int]:
"""各工作流的运行次数 {工作流 id: 次数},ids 为 None 时算全部。列表和单个工作流的接口都按它算:以前只有列表
计数,详情、保存、恢复的返回落在 WorkflowOut 的默认值 0 上,同一个工作流两处说的次数不一样。"""
stmt = select(Run.workflow_id, func.count(Run.id)).group_by(Run.workflow_id)
if ids is not None:
stmt = stmt.where(Run.workflow_id.in_(ids))
return dict((await session.execute(stmt)).all())
async def _workflow_out(session: AsyncSession, workflow: Workflow) -> WorkflowOut:
"""返回单个工作流的接口都走这里,run_count 和列表一致。"""
item = WorkflowOut.model_validate(workflow)
item.run_count = (await _run_counts(session, [workflow.id])).get(workflow.id, 0)
return item
@router.get("", response_model=list[WorkflowOut])
async def list_workflows(session: AsyncSession = Depends(get_session)) -> list[WorkflowOut]:
rows = list(
(await session.execute(select(Workflow).order_by(Workflow.updated_at.desc()))).scalars()
)
counts = dict(
(await session.execute(select(Run.workflow_id, func.count(Run.id)).group_by(Run.workflow_id))).all()
)
counts = await _run_counts(session)
out: list[WorkflowOut] = []
for row in rows:
item = WorkflowOut.model_validate(row)
@@ -84,7 +98,7 @@ async def list_workflows(session: AsyncSession = Depends(get_session)) -> list[W
@router.post("", response_model=WorkflowOut, status_code=201)
async def create_workflow(
payload: WorkflowIn, session: AsyncSession = Depends(get_session)
) -> Workflow:
) -> WorkflowOut:
workflow = Workflow(
name=payload.name,
description=payload.description,
@@ -99,23 +113,23 @@ async def create_workflow(
)
await session.commit()
await session.refresh(workflow)
return workflow
return await _workflow_out(session, workflow)
@router.get("/{workflow_id}", response_model=WorkflowOut)
async def get_workflow(
workflow_id: str, session: AsyncSession = Depends(get_session)
) -> Workflow:
) -> WorkflowOut:
workflow = await session.get(Workflow, workflow_id)
if not workflow:
raise HTTPException(404, _GONE)
return workflow
return await _workflow_out(session, workflow)
@router.patch("/{workflow_id}", response_model=WorkflowOut)
async def update_workflow(
workflow_id: str, payload: WorkflowPatch, session: AsyncSession = Depends(get_session)
) -> Workflow:
) -> WorkflowOut:
workflow = await session.get(Workflow, workflow_id)
if not workflow:
raise HTTPException(404, _GONE)
@@ -146,7 +160,7 @@ async def update_workflow(
workflow.status = "draft"
await session.commit()
await session.refresh(workflow)
return workflow
return await _workflow_out(session, workflow)
@router.delete("/{workflow_id}", status_code=204)
@@ -172,7 +186,7 @@ async def delete_workflow(
@router.post("/{workflow_id}/duplicate", response_model=WorkflowOut, status_code=201)
async def duplicate_workflow(
workflow_id: str, session: AsyncSession = Depends(get_session)
) -> Workflow:
) -> WorkflowOut:
source = await session.get(Workflow, workflow_id)
if not source:
raise HTTPException(404, _GONE)
@@ -185,7 +199,7 @@ async def duplicate_workflow(
session.add(copy)
await session.commit()
await session.refresh(copy)
return copy
return await _workflow_out(session, copy)
class VersionOut(BaseModel):
@@ -267,7 +281,7 @@ async def get_version(
@router.post("/{workflow_id}/versions/{version}/restore", response_model=WorkflowOut)
async def restore_version(
workflow_id: str, version: int, session: AsyncSession = Depends(get_session)
) -> Workflow:
) -> WorkflowOut:
workflow = await session.get(Workflow, workflow_id)
if not workflow:
raise HTTPException(404, _GONE)
@@ -298,7 +312,7 @@ async def restore_version(
workflow.status = "draft"
await session.commit()
await session.refresh(workflow)
return workflow
return await _workflow_out(session, workflow)
class PublishIn(BaseModel):
+84
View File
@@ -0,0 +1,84 @@
"""单个工作流接口的 run_count 和列表接口按同一个函数算。
2026-10 验收:GET /api/workflows/{id} 的 run_count 恒为 0,同一时刻列表接口(GET /api/workflows)里这个工作流是
2 次、运行列表按 workflow_id 筛出来也是 2 条——只有列表接口按 Run.workflow_id 计数,详情、保存(PATCH)、恢复旧版
直接返回数据库对象,WorkflowOut.run_count 落在默认值 0 上。现在凡是返回工作流的接口都按同一个函数算,次数和列表一致。
工作流名、图、运行都是虚构的。
"""
from __future__ import annotations
import pytest
from httpx import ASGITransport, AsyncClient
from app.db.base import SessionLocal
from app.db.models import Run
from app.main import app
GRAPH = {
"nodes": [
{"id": "start", "type": "input", "position": {"x": 0, "y": 0},
"data": {"label": "输入", "config": {"fields": [{"name": "week"}]}}},
{"id": "out", "type": "output", "position": {"x": 300, "y": 0},
"data": {"label": "成果", "config": {"fields": [{"name": "r", "value": "{{ input.week }}"}]}}},
],
"edges": [{"source": "start", "target": "out"}],
}
@pytest.fixture
async def client():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
yield c
async def _add_runs(workflow_id: str | None, n: int, status: str = "succeeded") -> None:
async with SessionLocal() as session:
for _ in range(n):
session.add(Run(workflow_id=workflow_id, workflow_name="灯塔周报", graph=GRAPH, input={}, status=status))
await session.commit()
async def _listed(client, workflow_id: str) -> int:
[row] = [w for w in (await client.get("/api/workflows")).json() if w["id"] == workflow_id]
return row["run_count"]
async def test_the_single_workflow_counts_its_runs_like_the_list(client):
made = (await client.post("/api/workflows", json={"name": "灯塔周报", "graph": GRAPH})).json()
other = (await client.post("/api/workflows", json={"name": "潮汐日报", "graph": GRAPH})).json()
await _add_runs(made["id"], 3)
await _add_runs(made["id"], 1, status="failed")
# 别的工作流、没挂在工作流上的运行不算进来
await _add_runs(other["id"], 2)
await _add_runs(None, 1)
assert await _listed(client, made["id"]) == 4
got = await client.get(f"/api/workflows/{made['id']}")
assert got.status_code == 200 and got.json()["run_count"] == 4
assert (await client.get(f"/api/workflows/{other['id']}")).json()["run_count"] == await _listed(client, other["id"]) == 2
async def test_save_and_restore_answer_with_the_same_count(client):
"""保存、恢复旧版的返回也是这个工作流:界面拿它替换手里的那一份,次数不能变回 0。"""
made = (await client.post("/api/workflows", json={"name": "灯塔周报", "graph": GRAPH})).json()
await _add_runs(made["id"], 2)
renamed = await client.patch(f"/api/workflows/{made['id']}", json={"name": "灯塔周报(新)"})
assert renamed.status_code == 200 and renamed.json()["run_count"] == 2
changed = {**GRAPH, "nodes": [*GRAPH["nodes"][:1], {**GRAPH["nodes"][1], "position": {"x": 320, "y": 0}}]}
saved = await client.patch(f"/api/workflows/{made['id']}", json={"graph": changed, "note": "挪了一下"})
assert saved.status_code == 200 and saved.json()["version"] == 2 and saved.json()["run_count"] == 2
restored = await client.post(f"/api/workflows/{made['id']}/versions/1/restore")
assert restored.status_code == 200 and restored.json()["run_count"] == 2
assert await _listed(client, made["id"]) == 2
async def test_new_and_copied_workflows_have_no_runs_yet(client):
"""新建、复制出来的是新工作流,还没有运行;复制不带上原来那个的次数。"""
made = (await client.post("/api/workflows", json={"name": "灯塔周报", "graph": GRAPH})).json()
assert made["run_count"] == 0
await _add_runs(made["id"], 3)
copy = await client.post(f"/api/workflows/{made['id']}/duplicate")
assert copy.status_code == 201 and copy.json()["run_count"] == 0
assert (await client.get(f"/api/workflows/{made['id']}")).json()["run_count"] == 3