158 lines
5.1 KiB
Python
158 lines
5.1 KiB
Python
"""用户端应用路由。
|
|
|
|
面向普通用户暴露“应用中心”能力：列出已发布工作流、上传输入创建任务、
|
|
查询进度、重试失败任务以及下载产物。用户只看到输入 -> 进度 -> 结果。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
|
|
from fastapi.responses import FileResponse
|
|
|
|
from app.db import Database
|
|
|
|
router = APIRouter(tags=["apps"])
|
|
|
|
|
|
def _now_iso() -> str:
|
|
"""返回当前 UTC 时间的 ISO 格式字符串。"""
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def _get_db() -> Database:
|
|
"""从 FastAPI 应用状态中延迟获取数据库实例。"""
|
|
from app.main import app
|
|
|
|
return app.state.db
|
|
|
|
|
|
@router.get("/api/apps")
|
|
def list_apps(db: Database = Depends(_get_db)) -> list[dict]:
|
|
"""返回全部已发布工作流及其最新版本定义。"""
|
|
apps = []
|
|
for workflow in db.list_workflows():
|
|
# 草稿工作流不对用户端可见。
|
|
if not workflow["published"]:
|
|
continue
|
|
latest = db.get_latest_workflow_version(workflow["id"])
|
|
apps.append(
|
|
{
|
|
"id": workflow["id"],
|
|
"name": workflow["name"],
|
|
"description": workflow["description"],
|
|
"version": workflow["latest_version"],
|
|
"definition": latest["definition"] if latest else None,
|
|
}
|
|
)
|
|
return apps
|
|
|
|
|
|
@router.post("/api/apps/{workflow_id}/runs")
|
|
async def create_run(
|
|
workflow_id: str,
|
|
file: UploadFile = File(...),
|
|
db: Database = Depends(_get_db),
|
|
) -> dict:
|
|
"""接收用户上传文件，创建排队中的工作流任务。"""
|
|
workflow = db.get_workflow(workflow_id)
|
|
# 只允许对已发布且存在版本的工作流发起任务。
|
|
if workflow is None or not workflow["published"]:
|
|
raise HTTPException(status_code=404, detail="published workflow not found")
|
|
|
|
latest = db.get_latest_workflow_version(workflow_id)
|
|
if latest is None:
|
|
raise HTTPException(status_code=422, detail="workflow has no version")
|
|
|
|
run_id = f"run_{uuid.uuid4().hex[:12]}"
|
|
# 使用安全文件名，避免路径穿越。
|
|
filename = Path(file.filename or "upload.bin").name
|
|
from app.config import STORAGE_DIR
|
|
|
|
# 上传文件按 run 隔离存放，调度器通过 input_uri 引用。
|
|
input_dir = STORAGE_DIR / "uploads" / run_id
|
|
input_dir.mkdir(parents=True, exist_ok=True)
|
|
input_uri = input_dir / filename
|
|
content = await file.read()
|
|
input_uri.write_bytes(content)
|
|
|
|
now = _now_iso()
|
|
db.create_run(
|
|
{
|
|
"id": run_id,
|
|
"workflow_id": workflow_id,
|
|
"workflow_version": latest["version"],
|
|
"status": "QUEUED",
|
|
"progress": 0,
|
|
"input_uri": str(input_uri),
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
}
|
|
)
|
|
return {
|
|
"id": run_id,
|
|
"status": "QUEUED",
|
|
"progress": 0,
|
|
"artifacts": [],
|
|
}
|
|
|
|
|
|
@router.get("/api/runs")
|
|
def list_runs(db: Database = Depends(_get_db)) -> list[dict]:
|
|
"""返回最近的运行记录，供任务管理页展示。"""
|
|
return db.list_runs()
|
|
|
|
|
|
@router.get("/api/runs/{run_id}")
|
|
def get_run(run_id: str, db: Database = Depends(_get_db)) -> dict:
|
|
"""返回任务详情，并附带当前产物列表。"""
|
|
run = db.get_run(run_id)
|
|
if run is None:
|
|
raise HTTPException(status_code=404, detail="run not found")
|
|
run["artifacts"] = db.list_artifacts(run_id)
|
|
return run
|
|
|
|
|
|
@router.post("/api/runs/{run_id}/retry")
|
|
def retry_run(run_id: str, db: Database = Depends(_get_db)) -> dict:
|
|
"""重置失败任务为排队状态，清空旧产物后重新执行。"""
|
|
run = db.get_run(run_id)
|
|
if run is None:
|
|
raise HTTPException(status_code=404, detail="run not found")
|
|
if run["status"] != "FAILED":
|
|
raise HTTPException(status_code=422, detail="only failed runs can be retried")
|
|
# reset_run 会清空进度、错误和旧产物，确保从头开始。
|
|
db.reset_run(run_id, _now_iso())
|
|
return {"id": run_id, "status": "QUEUED"}
|
|
|
|
|
|
@router.get("/api/runs/{run_id}/artifacts")
|
|
def list_run_artifacts(run_id: str, db: Database = Depends(_get_db)) -> list[dict]:
|
|
"""返回任务全部产物记录。"""
|
|
if db.get_run(run_id) is None:
|
|
raise HTTPException(status_code=404, detail="run not found")
|
|
return db.list_artifacts(run_id)
|
|
|
|
|
|
@router.get("/api/runs/{run_id}/artifacts/{artifact_name}")
|
|
def download_artifact(
|
|
run_id: str,
|
|
artifact_name: str,
|
|
db: Database = Depends(_get_db),
|
|
) -> FileResponse:
|
|
"""按任务与产物名下载文件，文件缺失时返回 404。"""
|
|
artifact = db.get_artifact(run_id, artifact_name)
|
|
if artifact is None:
|
|
raise HTTPException(status_code=404, detail="artifact not found")
|
|
path = Path(artifact["uri"])
|
|
if not path.is_file():
|
|
raise HTTPException(status_code=404, detail="artifact file missing")
|
|
return FileResponse(
|
|
path,
|
|
media_type=artifact["mime_type"],
|
|
filename=path.name,
|
|
)
|