Files
patbond-ai/app/api/routes.py
T
lixi 062e11b1ff
CI / ai-gates (push) Successful in 39s
feat(ai): 内部 HTTP 端点 + 服务票校验 + 契约漂移守卫(166 测试)
按 openapi-internal.yaml 落地两个端点与鉴权,全部离线可测。

**鉴权(app/api/auth.py)**是这一层的安全边界,三条纪律各有测试:
- **强制校验 aud 与 iss**(交给 pyjwt 强制,不解出来自己比——自己比容易漏掉
  「claim 不存在」那一路)。这是独立密钥对方案唯一的意义所在:Java 侧现有的
  JwtVerifier 只验签名、不校验 aud,正因如此才必须独立密钥;若本服务也只验签名,
  那对独立密钥就白搭了。
- **额外强制 typ=service**,作为「有人误把用户密钥配到这里」的第二道防线。
- **缺公钥时全拒而非全过**。配置缺失必须立刻暴露,而不是变成一扇敞开的门;
  与 Java 侧 InternalAuthFilter 未配置 internal-token 时全拒 /internal/** 一致。
- 失败一律返回同一个笼统原因,**不告诉调用方是签名错/过期/aud 不符**——对方是
  我们自己的服务,真正原因在日志里;细节回给调用方只会在将来某天变成提示。

**三类失败分开**(契约核心)。第三类「任务失败」走 200 + state=failed + failure,
因为那种情况下 HTTP 调用是成功的;做成非 2xx 会让「查不到任务」与「任务失败了」
难以区分。另外两处刻意选择:
- 服务商 429 **不透传**,一律 502 + provider 信封。429 既可能是「等几秒」也可能是
  「等到明天」,状态码表达不了这个区别,retry 才是权威信号。
- 超出模型限制的请求归 service/invalid_request 而非 provider——请求**根本没发出去**,
  归到 provider 会让调用方以为是上游问题而去退避重试,而重试同样的参数永远不会变好。
- 服务商 404(含超出其 7 天保留期)归 service/task_not_found:调用方要做的是
  **停止轮询**,不是退避重试。
- 接管 FastAPI 默认的 {"detail": [...]} 校验错误,否则调用方要处理三种形状,
  而第三种还是框架实现细节。

**契约漂移守卫(tests/api/test_contract_drift.py,14 项)**:本项目开工审计已证明
「没有门禁的契约必然漂移」——四模块快照锁实际不存在,白名单 42/41 之误四份文档
一路不可见整整一个迭代。内部契约同样需要门禁。比对路径/方法/字段集/枚举/必填项,
**不比对字节**(永远不会相同,只会制造一个天天红然后被 skip 绕过的门禁)。

写这个守卫时它立刻抓到一处问题,但**问题在我的内省方式而不在应用**:本版本
FastAPI 的 include_router 会包一层内部 wrapper,路由并不平铺在 app.routes 上。
已改用 app.openapi()——它是「究竟暴露了什么」的权威视图,且不依赖框架内部结构,
否则这个门禁会在升级框架时莫名失效,而一个悄悄失效的门禁比没有门禁更糟。

**新增 pyproject.toml**:`Depends()` 放在参数默认值里正是 FastAPI 的标准写法,
ruff 的 B008 在此是误报。按 ruff 正规方式声明这些调用可安全求值,而不是在每个
端点上撒 noqa——后者迟早被用来压制真正的问题。

schemas 用 extra="forbid":拼错的字段应当立刻报错而非被静默忽略,后者表现为
「我传了 durationSec 却没生效」这类极难查的问题。测试覆盖了 camelCase 与
user_id(契约刻意不含它)两种被拒情形。

新增依赖 fastapi / uvicorn / pyjwt[crypto];测试 127 → 166。

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-20 16:59:44 +08:00

109 lines
4.4 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.
"""内部接口路由。
**本层只做编排,不含任何厂商知识**:解析请求 → 按 `media_kind` 找适配器 → 调用 →
把领域结果序列化。厂商差异全在 `app/providers/<厂商>/` 内。
一处刻意的选择:路径参数是**服务商的任务 ID**,不是我方 `job_id`。本服务不持有
状态,唯一能用来问服务商的就是这个 ID;若用 `job_id` 就得存一份映射,「无状态」
不再成立,且会多出一个「Java 拿到 ID 但查不到任何东西」的中间态。
"""
from __future__ import annotations
import logging
from fastapi import APIRouter, Depends, Request
from app.domain.generation import MediaKind
from app.providers.base import GenerationProvider
from app.providers.registry import ProviderRegistry
from app.providers.volcengine.client import ArkError
from .auth import verify_service_token
from .errors import ProviderError, ServiceError, ServiceErrorCode
from .schemas import SubmitRequest, SubmitResponse, TaskResponse
log = logging.getLogger(__name__)
router = APIRouter(
prefix="/internal/generation",
dependencies=[Depends(verify_service_token)],
)
def _registry(request: Request) -> ProviderRegistry:
return request.app.state.registry
def _provider_for(
registry: ProviderRegistry, media_kind: MediaKind
) -> GenerationProvider:
"""按 `media_kind` 取适配器。
查不到即 `provider_unavailable`(503)而不是 500:这通常是凭证未配置导致适配器
没装配,属环境问题而非代码缺陷,且对调用方而言是**可重试**的——配好就能用。
"""
for provider in registry.all():
if provider.media_kind is media_kind:
return provider
raise ServiceError(
ServiceErrorCode.PROVIDER_UNAVAILABLE,
f"没有可用的 {media_kind.value} 生成适配器(检查服务商凭证是否已配置)",
)
@router.post("/tasks", response_model=SubmitResponse)
def submit_task(payload: SubmitRequest, request: Request) -> SubmitResponse:
"""提交生成任务,返回服务商任务 ID。
调用方**必须在上游产生任何计费动作之前**把该 ID 落库——火山不支持幂等键,
这是提交超时后唯一能判断「是否已提交」的依据。
"""
provider = _provider_for(_registry(request), payload.media_kind)
try:
provider_request_id = provider.submit(payload.to_domain())
except ArkError as exc:
# 服务商调用失败:原样带上分类,调用方据 retry 决定重试。
raise ProviderError(exc.failure) from exc
except ValueError as exc:
# 本地预校验拦下的参数问题(超出该模型的时长/分辨率/ratio 限制,或内容为空)。
# 属 invalid_request 而非 provider 错误——请求根本没发出去。
raise ServiceError(ServiceErrorCode.INVALID_REQUEST, str(exc)) from exc
log.info(
"已提交生成任务 job_id=%s provider_request_id=%s",
payload.job_id,
provider_request_id,
)
return SubmitResponse(provider_request_id=provider_request_id)
@router.get("/tasks/{provider_request_id}", response_model=TaskResponse)
def get_task(
provider_request_id: str,
request: Request,
media_kind: MediaKind = MediaKind.VIDEO,
) -> TaskResponse:
"""查询任务。
**任务失败也返回 200**:HTTP 调用成功了,失败的是任务,`state=failed` +
`failure` 表达它。做成非 2xx 会让「查不到任务」与「任务失败了」难以区分。
:param media_kind: 决定问哪个适配器。默认 video——当前只有视频适配器;
接入图片后调用方应显式传入。刻意不做「遍历所有适配器试一遍」:那会把
一个任务 ID 拿去问不相干的服务商。
"""
provider = _provider_for(_registry(request), media_kind)
try:
result = provider.fetch(provider_request_id)
except ArkError as exc:
# 服务商说任务不存在(含超过 7 天保留期)→ 这是 service 类的 task_not_found,
# 而不是 provider 错误:调用方要做的是停止轮询,不是退避重试。
if exc.http_status == 404:
raise ServiceError(
ServiceErrorCode.TASK_NOT_FOUND,
f"服务商无此任务(或已超出其 7 天保留期):{provider_request_id}",
) from exc
raise ProviderError(exc.failure) from exc
return TaskResponse.of(result)