3.5 KiB
3.5 KiB
FastAPI + Temporal MVP 图片流水线
这是一个最小可运行的图片生产流水线 MVP,使用 FastAPI + Temporal + SQLite + SQLAlchemy 实现,当前所有图像处理步骤都为 mock。
功能范围
- 低端全自动流程
auto_basic - 中端半自动流程
semi_pro - 创建订单、查询订单、查询资产、提交审核、查询 workflow 状态接口
- 中端流程支持
review signal - 中端流程支持
rerun_face / rerun_fusion / rerun_scene - 提供
.env.example、Alembic 初始迁移、基础测试
环境要求
- Python 3.11+
- Temporal CLI 或可用的 Temporal Server
安装依赖
python -m venv .venv
.venv\Scripts\activate
python -m pip install -U pip
python -m pip install -e .
环境变量
复制示例环境变量:
copy .env.example .env
默认数据库:
sqlite+aiosqlite:///./temporal_demo.db
启动 Temporal Server
temporal server start-dev
初始化数据库
推荐先执行 Alembic 迁移:
alembic upgrade head
如果没有先迁移,API 启动时也会在 AUTO_CREATE_TABLES=true 下自动建表,方便本地 MVP 调试。
启动 FastAPI
uvicorn app.main:app --reload
健康检查:
curl http://127.0.0.1:8000/healthz
启动 Worker
python -m app.workers.runner
API 调用示例
创建低端订单
curl -X POST http://127.0.0.1:8000/api/v1/orders ^
-H "Content-Type: application/json" ^
-d "{\"customer_level\":\"low\",\"service_mode\":\"auto_basic\",\"model_id\":101,\"pose_id\":3,\"garment_asset_id\":9001,\"scene_ref_asset_id\":8001}"
创建中端订单
curl -X POST http://127.0.0.1:8000/api/v1/orders ^
-H "Content-Type: application/json" ^
-d "{\"customer_level\":\"mid\",\"service_mode\":\"semi_pro\",\"model_id\":101,\"pose_id\":3,\"garment_asset_id\":9001,\"scene_ref_asset_id\":8001}"
查询订单详情
curl http://127.0.0.1:8000/api/v1/orders/1
查询订单资产
curl http://127.0.0.1:8000/api/v1/orders/1/assets
查询待审核列表
curl http://127.0.0.1:8000/api/v1/reviews/pending
提交审核通过
curl -X POST http://127.0.0.1:8000/api/v1/reviews/1/submit ^
-H "Content-Type: application/json" ^
-d "{\"decision\":\"approve\",\"reviewer_id\":77,\"comment\":\"通过\"}"
提交 rerun_face
curl -X POST http://127.0.0.1:8000/api/v1/reviews/1/submit ^
-H "Content-Type: application/json" ^
-d "{\"decision\":\"rerun_face\",\"reviewer_id\":77,\"comment\":\"面部不自然,重跑 face\"}"
查询 workflow 状态
curl http://127.0.0.1:8000/api/v1/workflows/1
流程说明
低端 auto_basic
prepare_modeltryonsceneqcexport
中端 semi_pro
prepare_modeltryonscenetexturefacefusionqc- 进入
waiting_review approve后进入exportrerun_scene回到scenererun_face回到facererun_fusion回到fusion
测试
pytest
说明:
- 首次运行测试时,
temporalio会自动下载 Temporal test server 二进制,需要可用外网;下载完成后会复用本地缓存。 - 如果是在全新环境中安装依赖,请优先使用
python -m pip install -e .,确保greenlet等运行时依赖一并安装。
覆盖范围:
- 健康检查
- 低端流程跑通
- 中端流程进入
waiting_review - 中端流程审核通过
- 中端流程 rerun 分支回流