后端

FastAPI 接口设计与数据一致性

交易和运营系统的接口难点通常不是把数据写进数据库,而是让不同入口遵守同一套规则:管理端看到的校验、公开预订接口返回的错误,以及并发请求下的最终数据,都应一致。

交易和运营系统的接口难点通常不是把数据写进数据库,而是让不同入口遵守同一套规则:管理端看到的校验、公开预订接口返回的错误,以及并发请求下的最终数据,都应一致。

先划清职责

我通常把一个写入接口拆成三层来思考:

层负责什么不应承担什么
Router解析请求、注入依赖、声明响应和状态码拼接复杂查询或决定业务规则
Service校验业务条件、组织事务、定义冲突语义依赖 HTTP 请求对象保存状态
Repository查询和持久化数据决定面向用户的错误文案

这不是要求每个接口都建立三个文件。简单查询可以保持简单;当一条规则被多个入口复用,或涉及多次写入时,再把规则集中到 Service。

把契约写在边界上

请求模型约束输入,响应模型限制输出。数据库模型可以包含内部状态,但这些字段不应因为直接返回 ORM 对象而意外暴露。

from uuid import UUID

from pydantic import BaseModel, Field


class LabelCreate(BaseModel):
    name: str = Field(min_length=1, max_length=80)


class LabelRead(BaseModel):
    id: UUID
    name: str

还要约定错误的含义。例如,同一公司下标签重名是 409 Conflict,请求字段缺失或格式不合法是 422 Unprocessable Entity。前端可以提前提示,但最终判断必须由服务端完成。租户或公司范围应从已验证的权限上下文取得,不能只相信请求体传来的 company_id。

并发下的唯一性

“先查询有没有重名,再插入”只能改善提示,不能保证唯一。两个请求可能同时查到不存在,然后同时写入。若归一化规则能稳定地由数据库表达,唯一索引是最直接的最终约束:

CREATE UNIQUE INDEX uq_label_company_name
ON booking_labels (company_id, normalized_name);

这段 SQL 是一种可选设计,并非下文实践案例的实现。实际系统若需要沿用应用层的 strip().casefold() 规则、兼容历史数据,也可以在同一事务中按公司取得锁,再检查所有相关记录并写入。此时所有创建、改名和其他写入入口都必须遵守同一把锁;它不像数据库唯一索引那样能保护绕过应用的写入。

无论选择哪种方式,冲突都应转换成稳定的 409 业务错误。不要把所有数据库异常都翻译成“重名”;其他异常仍应作为真实故障记录和处理。跨端实例见 预订字段重名规则的跨端闭环。

批量更新应当原子化

批量修改排期、库存或预订字段时,先检查整批请求的权限、重复项和引用关系,再在一个事务中写入。任一项失败就回滚整批,避免前端看到“保存失败”,数据库却只更新了一部分。

需要同时考虑重试:用户重复点击或网络超时后重发请求,接口是覆盖同一目标状态,还是会再创建一批记录?创建型操作可以引入幂等键;更新型操作可使用版本号检测过期编辑。具体选择取决于业务语义,不能只靠前端禁用按钮。

测试真实的冲突路径

除了成功路径,至少覆盖这些情况:

  1. 同一公司内重复名称返回约定的冲突错误,不留下半成品数据。
  2. 不同公司允许相同名称,且不能读取或修改彼此的数据。
  3. 两个并发请求争用同一名称时,最终只有一个成功。
  4. 批量请求中任一项无效时,其他项也不落库。
  5. OpenAPI 中的请求、响应与前端使用的字段一致。

并发和事务测试需要真正支持目标约束的测试数据库。仅用 mock Repository 验证 Service 调用顺序,无法证明数据库竞争下的结果。

流式接口的另一类边界

AI 聊天一类接口采用 SSE 时,响应一旦开始发送,后续异常就不能再改写成普通 JSON 错误。因此应在开始流式响应前完成鉴权和可预见的参数校验;流中则使用明确的事件类型表达内容、完成和错误,并在客户端断开时停止上游任务。代理层还需要允许数据逐块发送,否则本地看似正常,上线后可能只在结束时一次性显示全文。

从服务到浏览器的完整验收步骤见 SSE 流式聊天的端到端检查。