Files
writer-work-flow/apps/api/ww_api/main.py
Yaojia Wang 29349dc7ee feat(api): C2 多章链 服务+端点+schema+checkpointer 接线
承 C1 链图(build_chain_graph),落地多章工作流链的 apps/api 壳:
- 3 端点 routers/chain.py:POST .../chains/{key}/run→202 ChainRunAccepted;
  POST .../chains/runs/{job_id}/resume→202;GET /jobs/{id} 复用。校验:
  count 1..50→422、未知 chain_key→404、resume 非 awaiting→409、无凭据→503。
- schemas/chain.py:ChainRunRequest/ChainRunAccepted/ChainResumeRequest
  (ConflictDecision 复用 schemas/projects)。
- services/chain_runner.py:run_chain_job 仿 run_job 壳自建独立 session 驱动链图
  (set_running→ainvoke→据 __interrupt__ 置 awaiting_input/done/failed);
  build_accept_op 在 apps/api 装配验收事务闭包注入图节点(守 #3/#4);
  token 不入 result/日志。
- services/chain_deps.py:get_checkpointer_factory(运行时 AsyncPostgresSaver
  上下文 / 测试 MemorySaver)。
- 零迁移(设计 §7):复用 jobs,新增 status="awaiting_input" + JobRepo.set_awaiting,
  awaiting 章经 result.awaiting_chapter;新错误码 ErrorCode.CONFLICT(409)。
- project_deps:build_chain_gateway/get_chain_gateway(按请求 tier writer/analyst/light
  分派——单档网关恒返该档会错路由 review/digest)+ get_digest_gateway_builder。

单测 apps/api/tests/test_chain.py 12 用例(mock 网关 + MemorySaver + fake session/
accept_op,无 DB/无网络/无真 LLM):run/resume→202、未知 key 404、count 越界 422、
resume 非 awaiting 409、run_chain_job 无冲突→done、冲突→awaiting→resume→done、
错误脱敏、accept_op 冲突缺判→CONFLICT_UNRESOLVED。

门禁绿:ruff/format 干净 · mypy 193 Success · alembic 无漂移 · pytest 600 passed。
守不变量 #1/#3/#4/#5/#9。唯一新增 DDL(langgraph 检查点表)= C3 迁移。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-23 17:12:53 +02:00

128 lines
5.1 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""FastAPI 应用入口ARCH §7"""
from __future__ import annotations
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from ww_config import get_settings
from ww_core.domain.job_repo import SqlJobRepo
from ww_db import get_sessionmaker
from ww_shared import AppError, ErrorBody, ErrorCode, ErrorEnvelope
from ww_api.logging_config import configure_logging, get_logger
from ww_api.middleware import REQUEST_ID_HEADER, request_id_middleware
from ww_api.routers import (
chain,
foreshadow,
generation,
health,
jobs,
kimi_oauth,
outline,
projects,
rules,
settings_providers,
style,
toolbox,
)
from ww_api.security.credentials import _fernet
from ww_api.services.project_deps import seed_stub_user
configure_logging()
log = get_logger("ww.api")
@asynccontextmanager
async def _lifespan(app: FastAPI) -> AsyncIterator[None]:
# 加密 key 启动校验:缺失/非法则**快速失败**(不带病启动接流量)。`_fernet` 缺/非法抛
# CredentialKeyError此处让其冒泡使进程启动失败P0-2而非首次凭据读写才 500。
_fernet(get_settings().credential_enc_key.get_secret_value())
# 幂等 seed 单用户 stub——所有 owner_id FK 依赖它(见 memory/gotchas
async with get_sessionmaker()() as session:
await seed_stub_user(session)
# zombie reaperM4-d / §7.4 缓解):进程重启会丢未跑完的 BackgroundTask把残留
# status=running 的 job 标 failed让用户看到失败可重试而非进度条永转。自 commit、幂等。
async with get_sessionmaker()() as session:
reaped = await SqlJobRepo(session).reap_zombies()
if reaped:
log.info("job_zombies_reaped", count=reaped)
yield
def create_app() -> FastAPI:
app = FastAPI(title="网文创作工作流 API", version="0.0.0", lifespan=_lifespan)
# CORS前端Next.js与 API 分端口/跨源,浏览器端 `api.POST/PUT` 需 CORS 放行
# RSC 服务端取数同源、无需 CORS故此前同进程测试从未暴露此缺口。原型单用户、
# 本地开发:放行 localhost:3000可经 env `CORS_ORIGINS`(逗号分隔)覆盖。
settings = get_settings()
# `allow_credentials=True` 与通配 origin `"*"` 并存会被浏览器拒(且语义危险)——启动断言
# cors_origins 不含通配强制显式白名单P2 收窄)。
assert "*" not in settings.cors_origins, (
"cors_origins 不能含通配 '*'(与 allow_credentials=True 不兼容);请配置显式来源白名单"
)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=True,
# 收窄方法/头白名单(避免 `*` 放行任意方法/头P2
allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
allow_headers=["content-type", "authorization", REQUEST_ID_HEADER],
)
app.middleware("http")(request_id_middleware)
@app.exception_handler(AppError)
async def _app_error_handler(request: Request, exc: AppError) -> JSONResponse:
request_id = getattr(request.state, "request_id", None)
log.warning("app_error", code=exc.code, message=exc.message)
envelope = ErrorEnvelope(
error=ErrorBody(
code=exc.code,
message=exc.message,
details=exc.details,
request_id=request_id,
)
)
return JSONResponse(status_code=exc.http_status, content=envelope.model_dump())
@app.exception_handler(Exception)
async def _unhandled_error_handler(request: Request, exc: Exception) -> JSONResponse:
# 兜底:任何未被 AppError 捕获的异常 → 统一 INTERNAL 信封(不回显原始异常给客户端),
# 但服务端记完整上下文 + request_id 便于端到端排查P0-2
request_id = getattr(request.state, "request_id", None)
log.error(
"unhandled_exception",
request_id=request_id,
error_type=type(exc).__name__,
exc_info=exc,
)
envelope = ErrorEnvelope(
error=ErrorBody(
code=ErrorCode.INTERNAL,
message="服务器内部错误",
request_id=request_id,
)
)
return JSONResponse(status_code=500, content=envelope.model_dump())
app.include_router(health.router)
app.include_router(jobs.router)
app.include_router(projects.router)
app.include_router(foreshadow.router)
app.include_router(outline.router)
app.include_router(rules.router)
app.include_router(style.router)
app.include_router(generation.router)
app.include_router(generation.skills_router)
app.include_router(toolbox.router)
app.include_router(chain.router)
app.include_router(settings_providers.router)
app.include_router(kimi_oauth.router)
return app
app = create_app()