Python 基础体系 · 第 82/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。

Python Web API 工程：契约、错误、分页、幂等、限流和版本

一个 Web API 不是“把 Python 函数暴露成 HTTP 接口”这么简单。

真正可维护的 API，至少需要回答六个问题：

客户端究竟可以发送什么，服务端保证返回什么？
参数错误、业务冲突、权限失败和服务器故障如何区分？
大量数据如何稳定、可重复地分页？
客户端超时重试时，如何避免同一个操作执行两次？
流量超过处理能力时，如何拒绝请求而不是拖垮系统？
接口发生变化时，旧客户端如何继续工作？

这六个问题分别对应契约、错误、分页、幂等、限流和版本。它们不是相互独立的功能，而是共同决定 API 的行为边界。

本文使用 Python 3.14 和 FastAPI 展示一套完整的实现思路。FastAPI 基于 ASGI 运行；ASGI 把请求表示为 scope 和事件流，HTTP 请求通常对应一个独立的 HTTP scope，而长轮询等场景可能让该 scope 持续到响应结束。(asgi.readthedocs.io)

一、先建立 API 的运行模型

一个请求从客户端到业务代码，通常经过如下路径：

sequenceDiagram
    participant C as Client
    participant P as Reverse Proxy
    participant A as ASGI Server
    participant M as Middleware
    participant D as Dependency
    participant R as Router
    participant S as Service
    participant DB as Database

    C->>P: HTTP Request
    P->>A: Forward Request
    A->>M: ASGI scope/events
    M->>D: Authentication / Rate Limit
    D-->>M: Accepted or Error
    M->>R: Route Matching
    R->>S: Validated Input
    S->>DB: Read / Write
    DB-->>S: Result
    S-->>R: Domain Result
    R-->>M: Response Model
    M-->>A: HTTP Response
    A-->>P: Response
    P-->>C: HTTP Response

这里有三个容易混淆的层次：

HTTP 层：方法、路径、状态码、请求头、响应头和消息体。
框架层：路由匹配、参数解析、依赖注入、响应序列化和异常处理。
业务层：订单是否存在、库存是否足够、支付是否重复等业务规则。

如果把业务错误直接写成字符串，把数据库模型直接返回给客户端，或者让每个路由自行决定分页和错误格式，API 会很快失去一致性。

二、API 契约：客户端和服务端共同遵守的边界

2.1 什么是契约

API 契约是对以下内容的明确约定：

C = (M, U, I, O, E, V)

其中：

$M$ ：允许的 HTTP 方法，例如 GET、POST、PUT。
$U$ ：资源路径，例如 /v1/orders/{order_id}。
$I$ ：输入，包括路径参数、查询参数、请求头和请求体。
$O$ ：正常输出，包括状态码、响应头和响应体。
$E$ ：错误输出，包括错误分类、状态码和错误结构。
$V$ ：版本规则，例如 /v1 或媒体类型版本。

一个契约不是“接口文档里的一个 URL”，而是客户端据此编写代码时所依赖的完整行为。

例如，下面的接口契约可以表示为：

POST /v1/orders

请求头：
  Idempotency-Key: string，必填

请求体：
  {
    "product_id": integer，必须大于 0，
    "quantity": integer，范围 1..100
  }

成功：
  HTTP 201
  {
    "id": integer,
    "product_id": integer,
    "quantity": integer,
    "status": "created"
  }

错误：
  HTTP 400：请求格式错误
  HTTP 409：幂等键冲突或业务状态冲突
  HTTP 422：字段校验失败
  HTTP 429：超过限流阈值

客户端真正依赖的是这些细节：

请求头是否必须存在；
缺少字段时是 400 还是 422；
创建成功是 200 还是 201；
错误响应是否始终具有相同字段；
重试时是否可以安全地重复发送请求。

2.2 输入契约和输出契约必须分离

不要直接把数据库实体作为响应。

数据库对象可能包含：

{
    "id": 1,
    "product_id": 10,
    "quantity": 2,
    "status": "created",
    "internal_cost": 3.2,
    "payment_token": "secret",
}

客户端只应该看到：

{
    "id": 1,
    "product_id": 10,
    "quantity": 2,
    "status": "created",
}

FastAPI 的响应模型可以校验返回值、生成 OpenAPI Schema，并过滤未声明的字段；如果应用返回的数据不符合响应模型，通常说明服务端代码本身产生了错误，而不是应该把非法结构交给客户端。(fastapi.tiangolo.com)

from pydantic import BaseModel, ConfigDict, Field


class CreateOrderIn(BaseModel):
    model_config = ConfigDict(extra="forbid")

    product_id: int = Field(gt=0)
    quantity: int = Field(ge=1, le=100)


class OrderOut(BaseModel):
    id: int
    product_id: int
    quantity: int
    status: str

CreateOrderIn 和 OrderOut 的职责不同：

CreateOrderIn 描述客户端可以提交什么；
OrderOut 描述服务端承诺返回什么；
extra="forbid" 防止客户端悄悄提交未定义字段；
Field(gt=0) 和 Field(ge=1, le=100) 把基础约束写进契约。

2.3 HTTP 方法语义不是装饰

HTTP 方法表达的是操作语义，而不只是路由装饰器。

方法	常见含义	是否幂等
`GET`	获取资源	是
`POST`	创建资源或执行命令	通常否
`PUT`	用请求体替换目标资源	是
`PATCH`	部分修改资源	取决于设计
`DELETE`	删除目标资源	是

RFC 9110 将“幂等”定义为：对同一目标重复执行多个相同请求，其预期服务器效果应当与执行一次相同；标准方法中，安全方法、PUT 和 DELETE 具有幂等语义，而 POST 默认不具有幂等语义。(rfc-editor.org)

需要区分：

幂等 ≠ 返回值完全相同
幂等 ≠ 请求只执行一次
幂等 = 资源的预期最终效果相同

例如：

DELETE /v1/orders/10

第一次可能返回 204，第二次可能返回 404，但只要删除后的资源状态仍然是“不存在”，它仍可以符合幂等语义。

三、错误设计：错误也是 API 的输出契约

3.1 先区分错误类型

HTTP API 至少要区分以下错误：

错误类型	例子	常见状态码
语法错误	JSON 不能解析	`400`
参数校验错误	`quantity=-1`	`422`
认证失败	没有有效身份	`401`
授权失败	没有访问权限	`403`
资源不存在	订单不存在	`404`
状态冲突	重复创建、版本冲突	`409`
请求过多	触发限流	`429`
服务端故障	数据库不可用	`500`
上游故障	支付服务失败	`502` 或 `503`

不能把所有错误都返回：

{
  "error": "failed"
}

因为客户端无法知道：

是否应该修改请求；
是否应该刷新 Token；
是否应该等待后重试；
是否已经成功，只是响应丢失。

3.2 统一错误结构

可以定义一个稳定的错误外壳：

{
  "error": {
    "code": "ORDER_NOT_FOUND",
    "message": "订单不存在",
    "request_id": "req_01J...",
    "details": null
  }
}

字段含义：

code：机器可识别的稳定代码；
message：给开发者或用户看的说明；
request_id：关联日志、链路和故障排查；
details：字段级错误或额外上下文。

message 可以修改措辞，code 则应尽量保持兼容。

3.3 FastAPI 的异常处理路径

FastAPI 支持通过抛出 HTTPException 立即终止当前请求；它是异常，不是应该被 return 的普通值。(fastapi.tiangolo.com)

from fastapi import HTTPException


def require_positive(value: int) -> int:
    if value <= 0:
        raise HTTPException(
            status_code=422,
            detail="value must be positive",
        )
    return value

但在较大的系统中，业务层不应该到处直接构造 HTTP 异常。更清晰的做法是：

class OrderNotFound(Exception):
    pass


class DuplicateIdempotencyKey(Exception):
    pass

然后在 HTTP 边界统一转换：

from fastapi import Request
from fastapi.responses import JSONResponse


def error_body(
    request: Request,
    code: str,
    message: str,
    details=None,
) -> dict:
    request_id = getattr(request.state, "request_id", "unknown")
    return {
        "error": {
            "code": code,
            "message": message,
            "request_id": request_id,
            "details": details,
        }
    }


@app.exception_handler(OrderNotFound)
async def order_not_found_handler(
    request: Request,
    exc: OrderNotFound,
):
    return JSONResponse(
        status_code=404,
        content=error_body(
            request,
            "ORDER_NOT_FOUND",
            "订单不存在",
        ),
    )

这样业务服务只表达：

raise OrderNotFound

而不是依赖 FastAPI 的 Request 和 JSONResponse。

3.4 参数校验错误和业务错误不是一回事

下面两种错误经常被混淆：

{
  "product_id": 10,
  "quantity": -1
}

这是输入违反字段约束，属于参数校验错误。

而：

{
  "product_id": 10,
  "quantity": 2
}

如果产品已经下架，则是业务规则失败，即使 JSON 和类型都正确，也不能创建订单。

FastAPI 会在请求数据无效时产生 RequestValidationError，可以通过异常处理器定制其输出。(fastapi.tiangolo.com)

from fastapi.exceptions import RequestValidationError


@app.exception_handler(RequestValidationError)
async def validation_error_handler(
    request: Request,
    exc: RequestValidationError,
):
    details = [
        {
            "location": list(item["loc"]),
            "message": item["msg"],
            "type": item["type"],
        }
        for item in exc.errors()
    ]

    return JSONResponse(
        status_code=422,
        content=error_body(
            request,
            "VALIDATION_ERROR",
            "请求参数校验失败",
            details,
        ),
    )

错误处理器的关键边界是：

可预期、可分类、可安全暴露的错误 → 映射为明确的 4xx/5xx
未知异常 → 记录完整堆栈，对外返回通用 500

不要把数据库异常、SQL 语句或内部堆栈直接返回给客户端。

四、分页：不是把列表切片这么简单

4.1 分页的目标

假设资源总数为 $N$ ，每页大小为 $L$ 。

分页需要保证：

单次响应不会返回过大的数据；
客户端可以继续获取下一页；
页之间尽量不重复、不丢失；
数据新增或删除时，结果行为可解释；
查询具有明确排序。

如果没有排序，数据库返回顺序不属于可靠契约：

SELECT * FROM orders LIMIT 20;

即使当前看起来顺序稳定，执行计划、索引或并发写入都可能改变结果。

4.2 Offset 分页

Offset 分页通常写成：

SELECT id, product_id, quantity, status
FROM orders
ORDER BY id
LIMIT :limit OFFSET :offset;

页码和偏移量的关系是：

offset = (page - 1) \times limit

例如：

page = 3
limit = 20
offset = (3 - 1) × 20 = 40

查询的是第 41 到第 60 条记录。

Offset 分页适合：

后台管理界面；
数据量较小；
用户需要直接跳转到第 N 页；
数据变化不频繁。

它的主要问题是：如果第一页读取后，有新数据插入到前面，第二页的 offset 可能发生漂移。

初始数据：

[1, 2, 3, 4, 5, 6]

第一页取两条：

page 1 = [1, 2]

此时插入一条排在最前面的记录 0：

[0, 1, 2, 3, 4, 5, 6]

第二页仍然使用 OFFSET 2：

page 2 = [2, 3]

记录 2 重复出现，而记录 6 可能在后续读取中被推迟。

4.3 Cursor 分页

Cursor 分页不表示“跳过多少条”，而表示“从哪个稳定位置之后继续”。

假设按 id ASC 排序，游标记录最后一条数据的 id：

SELECT id, product_id, quantity, status
FROM orders
WHERE id > :after_id
ORDER BY id ASC
LIMIT :limit_plus_one;

之所以查询 limit + 1 条，是为了判断是否存在下一页：

请求 limit = 3
实际查询 4 条

返回前 3 条
如果第 4 条存在，则 next_cursor 非空

完整推导：

数据：[1, 2, 3, 4, 5, 6]

第一次：
after_id = None
查询：[1, 2, 3, 4]
返回：[1, 2, 3]
next_cursor = 3

第二次：
after_id = 3
查询：[4, 5, 6]
返回：[4, 5, 6]
next_cursor = None

如果排序字段可能重复，应使用复合游标：

WHERE (created_at, id) > (:created_at, :id)
ORDER BY created_at ASC, id ASC

因为只使用 created_at 时，同一时间戳下的记录可能被跳过或重复。

4.4 Cursor 必须是服务端可验证的数据

不要直接把内部游标暴露为：

?after_id=123

这会让客户端依赖内部排序字段，也可能暴露业务信息。

可以将游标编码为 Base64：

import base64


def encode_cursor(order_id: int) -> str:
    raw = str(order_id).encode()
    return base64.urlsafe_b64encode(raw).decode()


def decode_cursor(value: str | None) -> int | None:
    if value is None:
        return None

    try:
        raw = base64.urlsafe_b64decode(value.encode())
        order_id = int(raw.decode())
    except (ValueError, UnicodeDecodeError, base64.binascii.Error):
        raise ValueError("invalid cursor")

    if order_id < 0:
        raise ValueError("invalid cursor")

    return order_id

这只是编码，不是加密。若游标包含租户、过滤条件或敏感信息，应使用签名或加密，并在服务端验证：

cursor = encode({
    "tenant_id": 7,
    "sort": "created_at,id",
    "filter": {"status": "created"},
    "position": ["2026-09-01T10:00:00Z", 123]
})

如果客户端拿着一个来自旧查询的游标，改变了过滤条件，服务端必须拒绝，或者明确规定其行为。否则同一个游标可能被错误地应用于不同数据集。

五、幂等：处理超时、重试和重复提交

5.1 幂等的真正问题

客户端发送请求后，可能发生以下情况：

sequenceDiagram
    participant C as Client
    participant API as API
    participant DB as Database

    C->>API: POST + Idempotency-Key: k1
    API->>DB: 创建订单
    DB-->>API: 提交成功
    API--x C: 响应在网络中丢失

    C->>API: 重试 POST + Idempotency-Key: k1
    API->>DB: 查询幂等记录
    DB-->>API: 已处理
    API-->>C: 返回第一次结果

如果没有幂等机制，第二次请求可能再次扣款、再次创建订单或再次发送邮件。

HTTP 标准只保证方法本身的幂等语义，不能自动让任意 POST 业务变得幂等。对于非幂等方法，客户端通常不应在无法判断原请求是否成功时自动重试，除非接口另有幂等设计。(rfc-editor.org)

5.2 Idempotency-Key 的契约

对于需要可靠重试的创建操作，可以要求：

POST /v1/orders
Idempotency-Key: 01JABC...
Content-Type: application/json

服务端保存：

(租户或用户, 幂等键)
    → 请求指纹
    → 状态
    → HTTP 状态码
    → 响应体
    → 过期时间

状态至少可以分为：

ABSENT
  ↓ 第一次请求
PROCESSING
  ↓ 业务成功
COMPLETED

PROCESSING
  ↓ 业务失败
FAILED 或删除记录

COMPLETED
  ↓ 重复请求
返回已保存响应

请求指纹用于防止同一个键被用于不同请求：

fingerprint = SHA256(method \Vert path \Vert canonical\_body)

如果：

第一次：
key = k1
body = {"product_id": 10, "quantity": 2}

第二次：
key = k1
body = {"product_id": 10, "quantity": 3}

则不能返回第一次结果，因为这会掩盖客户端错误。应返回 409 Conflict。

5.3 并发条件

下面的代码是不安全的：

if key not in records:
    records[key] = "processing"
    create_order()

两个并发请求可能同时执行：

请求 A：检查 key，不存在
请求 B：检查 key，不存在
请求 A：创建订单
请求 B：创建订单

必须让“检查并占用幂等键”具备原子性。

在单进程示例中，可以用 asyncio.Lock：

import asyncio


idempotency_lock = asyncio.Lock()
idempotency_records: dict[str, dict] = {}


async def claim_idempotency_key(
    key: str,
    fingerprint: str,
) -> dict | None:
    async with idempotency_lock:
        old = idempotency_records.get(key)

        if old is not None:
            if old["fingerprint"] != fingerprint:
                raise DuplicateIdempotencyKey

            if old["status"] == "completed":
                return old

            raise HTTPException(
                status_code=409,
                detail="request with this key is still processing",
            )

        idempotency_records[key] = {
            "fingerprint": fingerprint,
            "status": "processing",
        }
        return None

但这个锁只在当前 Python 进程内有效。如果服务启动了多个 worker，或者部署了多个实例：

Worker A 的内存锁 ≠ Worker B 的内存锁

生产环境应把幂等记录放入共享存储，并利用数据库唯一约束：

CREATE TABLE idempotency_keys (
    tenant_id     BIGINT NOT NULL,
    idem_key      VARCHAR(255) NOT NULL,
    fingerprint   CHAR(64) NOT NULL,
    status        VARCHAR(20) NOT NULL,
    status_code   INTEGER,
    response_body JSONB,
    expires_at    TIMESTAMP NOT NULL,
    PRIMARY KEY (tenant_id, idem_key)
);

事务流程应是：

1. 开启事务
2. INSERT 幂等记录
3. 如果唯一键冲突，则读取旧记录
4. 首次请求执行创建订单
5. 订单和幂等记录在同一个事务中提交
6. 后续请求直接返回保存的响应

关键是第 5 步：订单写入和幂等记录必须具有一致的提交边界。否则可能出现：

订单已创建，但幂等记录未保存

客户端重试后仍会再次创建订单。

六、限流：控制请求速率，而不是简单计数

6.1 限流解决什么问题

限流的目标不是惩罚客户端，而是限制某个主体消耗共享资源的速度。

主体可以是：

IP；
用户；
API Key；
租户；
路由；
全局服务。

最简单的固定窗口算法：

每分钟允许 100 次请求
当前时间窗口内计数 >= 100 → 返回 429
进入下一分钟 → 计数归零

它容易实现，但存在窗口边界突刺：

12:00:59：发送 100 次
12:01:00：再发送 100 次

两秒内实际通过 200 次请求

6.2 Token Bucket

令牌桶使用两个参数：

$r$ ：令牌生成速率，单位为令牌/秒；
$B$ ：桶容量；
$T$ ：上次更新时间；
$tokens$ ：当前令牌数。

当前时间为 $now$ 时：

tokens' = \min(B,\ tokens + (now - T) \times r)

如果请求成本为 $c$ ，则：

tokens' >= c → 扣除 c 个令牌，允许请求
tokens' <  c → 拒绝请求

例如：

r = 2 个/秒
B = 5
初始 tokens = 5

连续请求的状态：

第 1 次：5 → 4，允许
第 2 次：4 → 3，允许
第 3 次：3 → 2，允许
第 4 次：2 → 1，允许
第 5 次：1 → 0，允许
第 6 次：0 → 0，拒绝
等待 0.5 秒：0 → 1，允许

429 Too Many Requests 表示客户端在一段时间内发送了过多请求，响应可以携带 Retry-After 告知客户端等待时间。(rfc-editor.org)

6.3 单进程 Token Bucket 示例

import asyncio
import time
from dataclasses import dataclass


@dataclass
class Bucket:
    tokens: float
    updated_at: float


class InMemoryRateLimiter:
    def __init__(self, rate: float, capacity: float):
        self.rate = rate
        self.capacity = capacity
        self._buckets: dict[str, Bucket] = {}
        self._lock = asyncio.Lock()

    async def consume(self, key: str, cost: float = 1) -> float | None:
        now = time.monotonic()

        async with self._lock:
            bucket = self._buckets.get(
                key,
                Bucket(
                    tokens=self.capacity,
                    updated_at=now,
                ),
            )

            elapsed = max(0.0, now - bucket.updated_at)
            bucket.tokens = min(
                self.capacity,
                bucket.tokens + elapsed * self.rate,
            )
            bucket.updated_at = now

            if bucket.tokens < cost:
                missing = cost - bucket.tokens
                retry_after = missing / self.rate
                self._buckets[key] = bucket
                return retry_after

            bucket.tokens -= cost
            self._buckets[key] = bucket
            return None

依赖中使用：

from fastapi import Depends, Header, Request, Response


limiter = InMemoryRateLimiter(rate=2, capacity=5)


async def enforce_rate_limit(
    request: Request,
    response: Response,
    authorization: str | None = Header(default=None),
):
    identity = authorization or request.client.host or "anonymous"
    retry_after = await limiter.consume(identity)

    if retry_after is not None:
        seconds = max(1, int(retry_after + 0.999))
        response.headers["Retry-After"] = str(seconds)

        raise HTTPException(
            status_code=429,
            detail="rate limit exceeded",
            headers={"Retry-After": str(seconds)},
        )

FastAPI 的依赖注入可以执行共享逻辑，并将结果注入路径操作函数；依赖及其参数也会被纳入 OpenAPI。(fastapi.tiangolo.com)

但是这个实现有明确限制：

单进程有效
重启后计数丢失
内存会随 key 数量增长
多个 worker 不共享配额

生产环境通常把令牌桶状态放在 Redis 等共享存储中，并使用原子 Lua 脚本或等价机制完成“补充令牌、判断、扣减”。

6.4 限流键不能盲目使用客户端 IP

如果应用位于反向代理之后，request.client.host 可能只是代理地址。直接信任任意 X-Forwarded-For 也可能让客户端伪造 IP。

因此限流身份的优先级应由部署拓扑决定：

已认证用户 / API Key
    >
租户
    >
可信代理解析出的客户端 IP
    >
连接对端地址

限流还必须考虑重试风暴：

客户端收到 429
所有客户端立刻重试
服务端仍然持续返回 429

因此响应应包含 Retry-After，客户端也应使用指数退避和随机抖动，而不是固定时间同时重试。

七、版本：管理不可兼容的契约变化

7.1 什么变化算破坏兼容性

下面这些变化通常会破坏旧客户端：

删除响应字段
修改字段类型
把可选字段改为必填
改变字段含义
改变错误代码
改变分页游标规则
改变默认排序
改变认证要求

例如原契约：

{
  "id": 1,
  "status": "created"
}

如果新版本改为：

{
  "id": "1",
  "state": 1
}

这不是普通重构，而是客户端可观察行为发生了变化。

7.2 URL 版本

最直接的方案是路径版本：

/v1/orders
/v2/orders

FastAPI 可以用路由器组织不同版本：

from fastapi import APIRouter, FastAPI

app = FastAPI()

v1 = APIRouter(prefix="/v1")
v2 = APIRouter(prefix="/v2")


@v1.get("/orders/{order_id}", response_model=OrderOut)
async def get_order_v1(order_id: int):
    ...


@v2.get("/orders/{order_id}", response_model=OrderV2Out)
async def get_order_v2(order_id: int):
    ...


app.include_router(v1)
app.include_router(v2)

优点是：

路由清晰；
OpenAPI 文档可同时展示；
客户端容易选择版本；
灰度和回滚相对直接。

缺点是：

需要维护多个契约；
业务逻辑可能被重复实现；
版本长期不下线会增加测试矩阵。

较好的内部结构是：

HTTP v1 adapter ─┐
                  ├─> Domain Service ─> Repository
HTTP v2 adapter ─┘

版本差异放在输入转换、输出转换和错误适配层，不要复制整个业务系统。

7.3 兼容性策略

可以把变化分成三类：

可兼容增加

例如在响应中增加一个客户端可忽略的新字段：

{
  "id": 1,
  "status": "created",
  "created_at": "2026-09-01T10:00:00Z"
}

但前提是客户端遵循“未知字段可忽略”，而不是严格要求字段集合完全相等。

可兼容弃用

先保留旧字段，同时增加新字段：

{
  "status": "created",
  "state": "created"
}

然后通过文档、响应头或监控观察旧字段使用情况，最后再删除。

不兼容变化

如果必须改变字段类型、语义或分页规则，创建新版本：

/v1/orders
/v2/orders

版本号不能替代迁移策略。服务端仍需要：

明确弃用时间；
提供迁移文档；
记录客户端版本和调用量；
在下线前验证没有关键调用方；
保留回滚方案。

八、一个可运行的 FastAPI 端到端示例

下面的示例把几个机制组合起来：

/v1/orders；
Pydantic 输入和输出模型；
统一错误结构；
Cursor 分页；
Idempotency-Key；
进程内 Token Bucket；
应用生命周期初始化；
请求 ID 中间件。

它使用内存存储，因此适合理解流程，不适合作为多实例生产实现。

8.1 完整代码

保存为 main.py：

from __future__ import annotations

import asyncio
import base64
import hashlib
import json
import time
import uuid
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import Annotated, Any

from fastapi import (
    Depends,
    FastAPI,
    Header,
    HTTPException,
    Request,
    Response,
    status,
)
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel, ConfigDict, Field


# ---------- 数据模型 ----------

class CreateOrderIn(BaseModel):
    model_config = ConfigDict(extra="forbid")

    product_id: int = Field(gt=0)
    quantity: int = Field(ge=1, le=100)


class OrderOut(BaseModel):
    id: int
    product_id: int
    quantity: int
    status: str


class OrderPage(BaseModel):
    items: list[OrderOut]
    next_cursor: str | None = None


# ---------- 业务异常 ----------

class OrderNotFound(Exception):
    pass


class IdempotencyConflict(Exception):
    pass


# ---------- 应用状态 ----------

@dataclass
class Bucket:
    tokens: float
    updated_at: float


class AppState:
    def __init__(self):
        self.orders: dict[int, OrderOut] = {}
        self.next_order_id = 1

        self.idempotency: dict[str, dict[str, Any]] = {}
        self.idempotency_lock = asyncio.Lock()

        self.buckets: dict[str, Bucket] = {}
        self.rate_limit_lock = asyncio.Lock()


@asynccontextmanager
async def lifespan(app: FastAPI):
    state = AppState()

    # 这里可以初始化数据库连接池、Redis 客户端等共享资源。
    state.orders[1] = OrderOut(
        id=1,
        product_id=100,
        quantity=2,
        status="created",
    )
    state.next_order_id = 2

    app.state.api = state

    yield

    # 这里可以关闭连接池、刷新缓冲区、停止后台任务。
    app.state.api = None


app = FastAPI(
    title="Order API",
    version="1.0.0",
    lifespan=lifespan,
)


# ---------- 工具函数 ----------

def get_state(request: Request) -> AppState:
    return request.app.state.api


def make_error(
    request: Request,
    code: str,
    message: str,
    details: Any = None,
) -> dict[str, Any]:
    return {
        "error": {
            "code": code,
            "message": message,
            "request_id": getattr(
                request.state,
                "request_id",
                "unknown",
            ),
            "details": details,
        }
    }


def encode_cursor(order_id: int) -> str:
    raw = str(order_id).encode("ascii")
    return base64.urlsafe_b64encode(raw).decode("ascii")


def decode_cursor(cursor: str | None) -> int | None:
    if cursor is None:
        return None

    try:
        raw = base64.urlsafe_b64decode(cursor.encode("ascii"))
        value = int(raw.decode("ascii"))
    except (ValueError, UnicodeDecodeError, base64.binascii.Error):
        raise HTTPException(
            status_code=400,
            detail="invalid cursor",
        )

    if value < 0:
        raise HTTPException(
            status_code=400,
            detail="invalid cursor",
        )

    return value


def request_fingerprint(payload: CreateOrderIn) -> str:
    canonical = json.dumps(
        payload.model_dump(),
        sort_keys=True,
        separators=(",", ":"),
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


# ---------- 中间件 ----------

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = request.headers.get(
        "X-Request-ID",
        f"req_{uuid.uuid4().hex}",
    )
    request.state.request_id = request_id

    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response


# ---------- 限流依赖 ----------

async def enforce_rate_limit(
    request: Request,
    response: Response,
    state: Annotated[AppState, Depends(get_state)],
    authorization: Annotated[
        str | None,
        Header()
    ] = None,
):
    # 示例只使用请求头作为身份键。
    # 生产环境应使用已经验证过的用户、API Key 或租户身份。
    identity = authorization or request.client.host or "anonymous"

    rate = 2.0       # 每秒补充 2 个令牌
    capacity = 5.0   # 最多积累 5 个令牌
    now = time.monotonic()

    async with state.rate_limit_lock:
        bucket = state.buckets.get(
            identity,
            Bucket(tokens=capacity, updated_at=now),
        )

        elapsed = max(0.0, now - bucket.updated_at)
        bucket.tokens = min(
            capacity,
            bucket.tokens + elapsed * rate,
        )
        bucket.updated_at = now

        if bucket.tokens < 1:
            retry_after = max(
                1,
                int((1 - bucket.tokens) / rate + 0.999),
            )
            response.headers["Retry-After"] = str(retry_after)

            raise HTTPException(
                status_code=429,
                detail="rate limit exceeded",
                headers={"Retry-After": str(retry_after)},
            )

        bucket.tokens -= 1
        state.buckets[identity] = bucket


# ---------- 统一错误处理 ----------

@app.exception_handler(RequestValidationError)
async def request_validation_error_handler(
    request: Request,
    exc: RequestValidationError,
):
    details = [
        {
            "location": list(error["loc"]),
            "message": error["msg"],
            "type": error["type"],
        }
        for error in exc.errors()
    ]

    return JSONResponse(
        status_code=422,
        content=make_error(
            request,
            "VALIDATION_ERROR",
            "请求参数校验失败",
            details,
        ),
    )


@app.exception_handler(OrderNotFound)
async def order_not_found_handler(
    request: Request,
    exc: OrderNotFound,
):
    return JSONResponse(
        status_code=404,
        content=make_error(
            request,
            "ORDER_NOT_FOUND",
            "订单不存在",
        ),
    )


@app.exception_handler(IdempotencyConflict)
async def idempotency_conflict_handler(
    request: Request,
    exc: IdempotencyConflict,
):
    return JSONResponse(
        status_code=409,
        content=make_error(
            request,
            "IDEMPOTENCY_KEY_REUSED",
            "同一个幂等键不能对应不同请求",
        ),
    )


@app.exception_handler(HTTPException)
async def http_exception_handler(
    request: Request,
    exc: HTTPException,
):
    headers = exc.headers or {}

    return JSONResponse(
        status_code=exc.status_code,
        content=make_error(
            request,
            f"HTTP_{exc.status_code}",
            str(exc.detail),
        ),
        headers=headers,
    )


# ---------- API v1 ----------

@app.get(
    "/v1/orders/{order_id}",
    response_model=OrderOut,
    dependencies=[Depends(enforce_rate_limit)],
)
async def get_order(
    order_id: int,
    state: Annotated[AppState, Depends(get_state)],
):
    order = state.orders.get(order_id)

    if order is None:
        raise OrderNotFound

    return order


@app.get(
    "/v1/orders",
    response_model=OrderPage,
    dependencies=[Depends(enforce_rate_limit)],
)
async def list_orders(
    state: Annotated[AppState, Depends(get_state)],
    limit: int = 20,
    cursor: str | None = None,
):
    if not 1 <= limit <= 100:
        raise HTTPException(
            status_code=422,
            detail="limit must be between 1 and 100",
        )

    after_id = decode_cursor(cursor)

    orders = [
        order
        for order in sorted(
            state.orders.values(),
            key=lambda item: item.id,
        )
        if after_id is None or order.id > after_id
    ]

    selected = orders[: limit + 1]
    has_next = len(selected) > limit
    items = selected[:limit]

    next_cursor = (
        encode_cursor(items[-1].id)
        if has_next and items
        else None
    )

    return OrderPage(
        items=items,
        next_cursor=next_cursor,
    )


@app.post(
    "/v1/orders",
    response_model=OrderOut,
    status_code=status.HTTP_201_CREATED,
    dependencies=[Depends(enforce_rate_limit)],
)
async def create_order(
    payload: CreateOrderIn,
    state: Annotated[AppState, Depends(get_state)],
    idempotency_key: Annotated[
        str | None,
        Header(alias="Idempotency-Key")
    ] = None,
):
    if not idempotency_key:
        raise HTTPException(
            status_code=400,
            detail="Idempotency-Key is required",
        )

    fingerprint = request_fingerprint(payload)

    async with state.idempotency_lock:
        old = state.idempotency.get(idempotency_key)

        if old is not None:
            if old["fingerprint"] != fingerprint:
                raise IdempotencyConflict

            if old["status"] == "completed":
                return old["response"]

            raise HTTPException(
                status_code=409,
                detail="request is still processing",
            )

        state.idempotency[idempotency_key] = {
            "fingerprint": fingerprint,
            "status": "processing",
        }

        # 示例中用内存锁把“占用幂等键”和“创建订单”串起来。
        # 多进程生产环境必须改为数据库唯一约束和事务。
        order_id = state.next_order_id
        state.next_order_id += 1

        order = OrderOut(
            id=order_id,
            product_id=payload.product_id,
            quantity=payload.quantity,
            status="created",
        )
        state.orders[order_id] = order

        state.idempotency[idempotency_key] = {
            "fingerprint": fingerprint,
            "status": "completed",
            "response": order,
        }

        return order

8.2 启动

安装依赖：

python3.14 -m venv .venv
source .venv/bin/activate
python -m pip install "fastapi[standard]"

启动开发服务器：

fastapi dev main.py

FastAPI 的标准安装方式会带上常用的可选依赖；具体部署时仍应锁定依赖版本并使用生产级 ASGI Server 配置。(fastapi.tiangolo.com)

8.3 查询订单

curl -i http://127.0.0.1:8000/v1/orders/1

预期结果：

HTTP/1.1 200 OK
content-type: application/json
x-request-id: req_...

{
  "id": 1,
  "product_id": 100,
  "quantity": 2,
  "status": "created"
}

不存在的订单：

curl -i http://127.0.0.1:8000/v1/orders/999

预期结果：

HTTP/1.1 404 Not Found
content-type: application/json
x-request-id: req_...

{
  "error": {
    "code": "ORDER_NOT_FOUND",
    "message": "订单不存在",
    "request_id": "req_...",
    "details": null
  }
}

8.4 创建订单和重试

第一次请求：

curl -i \
  -X POST http://127.0.0.1:8000/v1/orders \
  -H 'Content-Type: application/json' \
  -H 'Idempotency-Key: order-request-001' \
  -d '{"product_id": 200, "quantity": 3}'

预期：

HTTP/1.1 201 Created

{
  "id": 2,
  "product_id": 200,
  "quantity": 3,
  "status": "created"
}

使用相同键重试：

curl -i \
  -X POST http://127.0.0.1:8000/v1/orders \
  -H 'Content-Type: application/json' \
  -H 'Idempotency-Key: order-request-001' \
  -d '{"product_id": 200, "quantity": 3}'

结果仍然是订单 2，而不是再次产生订单 3。

如果复用键但修改请求体：

curl -i \
  -X POST http://127.0.0.1:8000/v1/orders \
  -H 'Content-Type: application/json' \
  -H 'Idempotency-Key: order-request-001' \
  -d '{"product_id": 200, "quantity": 4}'

应返回：

HTTP/1.1 409 Conflict

{
  "error": {
    "code": "IDEMPOTENCY_KEY_REUSED",
    "message": "同一个幂等键不能对应不同请求",
    "request_id": "req_...",
    "details": null
  }
}

8.5 Cursor 分页

第一次请求：

curl -s 'http://127.0.0.1:8000/v1/orders?limit=1'

可能返回：

{
  "items": [
    {
      "id": 1,
      "product_id": 100,
      "quantity": 2,
      "status": "created"
    }
  ],
  "next_cursor": "MQ=="
}

第二次请求：

curl -s \
  'http://127.0.0.1:8000/v1/orders?limit=1&cursor=MQ=='

服务端先解码：

"MQ==" → "1" → after_id = 1

然后查询：

id > 1
ORDER BY id ASC
LIMIT 2

返回第一条记录，并根据是否存在第二条记录决定是否生成下一游标。

8.6 触发限流

令牌桶容量是 5，生成速率是每秒 2 个令牌。快速连续发送超过 5 个请求：

for i in 1 2 3 4 5 6 7; do
  curl -i -s http://127.0.0.1:8000/v1/orders/1 \
    | grep -E 'HTTP/|retry-after|rate limit|ORDER'
done

后续请求会得到：

HTTP/1.1 429 Too Many Requests
retry-after: 1

{
  "error": {
    "code": "HTTP_429",
    "message": "rate limit exceeded",
    "request_id": "req_...",
    "details": null
  }
}

这里的 Retry-After 不是“保证下一次一定成功”，而是服务端根据当前令牌不足量估算的最短等待时间。并发请求、多个实例和代理层限流都可能使实际结果不同。

九、生命周期、依赖和中间件的边界

9.1 应用生命周期

数据库连接池、Redis 客户端、模型加载器等资源通常属于应用级资源，不应为每个请求重复创建。

FastAPI 的 lifespan 逻辑在应用开始接收请求之前执行初始化，在应用停止后执行清理，适合管理跨请求共享资源。(fastapi.tiangolo.com)

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.db = await create_pool()
    try:
        yield
    finally:
        await app.state.db.close()

生命周期与请求依赖不同：

应用生命周期：
  启动一次 → 服务多个请求 → 关闭一次

请求依赖：
  每个请求执行 → 注入当前请求 → 请求结束清理

带 yield 的 FastAPI 依赖可以在 yield 前创建资源，在 finally 中清理资源；它适合数据库会话等请求级资源。(fastapi.tiangolo.com)

async def get_db():
    session = Session()
    try:
        yield session
    finally:
        session.close()

9.2 中间件和依赖的选择

中间件适合横切所有请求的逻辑：

请求 ID
访问日志
耗时统计
CORS
全局安全响应头

依赖适合某组路由共享、且需要参与参数解析或身份注入的逻辑：

当前用户
租户
权限
数据库会话
限流主体

FastAPI 中间件会在具体路径操作之前处理请求，在响应返回前再次处理响应；带 yield 的依赖清理时机和后台任务也有明确的执行顺序。(fastapi.tiangolo.com)

不要把所有逻辑都放入中间件：

@app.middleware("http")
async def everything(request, call_next):
    ...

这样会导致路由无法清楚声明自己的前置条件，也不利于生成契约和测试。

9.3 异步不等于自动并行

Python 3.14 的 asyncio 主要通过事件循环和协作式调度处理 I/O 并发，适合网络、文件、数据库等等待较多的场景；如果在 async def 中调用阻塞函数，整个事件循环线程可能被阻塞。(docs.python.org)

错误示例：

@app.get("/bad")
async def bad():
    time.sleep(5)
    return {"ok": True}

time.sleep(5) 会阻塞事件循环。

应使用异步 API：

@app.get("/good")
async def good():
    await asyncio.sleep(5)
    return {"ok": True}

或者把无法避免的阻塞工作移出事件循环：

import asyncio


def blocking_call():
    time.sleep(5)
    return "done"


@app.get("/bridge")
async def bridge():
    result = await asyncio.to_thread(blocking_call)
    return {"result": result}

这只适用于适合线程执行的阻塞 I/O。CPU 密集型计算仍可能占用线程，应考虑进程池、专用任务队列或独立计算服务。

十、生产边界：示例代码为什么不能直接上线

10.1 内存状态不适合多实例

本文示例把订单、幂等键和限流桶都放在内存中。部署两个 worker 后：

请求 1 → Worker A，创建订单
请求 2 → Worker B，看不到 A 的幂等记录

结果可能是重复创建。

生产实现通常需要：

订单数据       → 数据库
幂等记录       → 数据库或共享 KV
限流状态       → Redis 等共享存储
异步任务       → 消息队列
请求日志       → 集中式日志系统

10.2 幂等记录必须考虑过期和失败恢复

幂等键不能永久保留，否则存储会无限增长。过期时间应依据业务风险决定：

支付类请求可能需要较长保存时间；
普通资源创建可能只需覆盖客户端重试窗口；
已完成请求和处理中请求可以采用不同策略。

如果服务在 PROCESSING 状态崩溃，后续请求不能永久得到“仍在处理中”。需要：

processing_started_at
lease_expiry
worker_owner

超过租约后，可以由新请求接管，或者进入人工恢复流程。

10.3 分页必须绑定排序和过滤条件

以下接口不完整：

GET /orders?cursor=abc

因为客户端不知道 abc 是基于什么生成的：

排序字段是什么？
升序还是降序？
过滤条件是否相同？
租户是否相同？

更安全的游标应该绑定上下文：

{
  "tenant_id": 7,
  "sort": "created_at,id",
  "direction": "asc",
  "filter": {
    "status": "created"
  },
  "position": {
    "created_at": "2026-09-01T10:00:00Z",
    "id": 123
  }
}

10.4 限流不是容量管理的全部

限流只能控制请求进入速度，不能解决：

数据库连接池耗尽；
下游服务变慢；
单请求执行时间过长；
内存持续增长；
消息队列堆积；
某个租户消耗全部共享资源。

因此还需要：

请求超时
数据库连接池上限
并发数限制
熔断
舱壁隔离
队列长度上限
健康检查

如果下游服务已经不可用，继续接受大量请求只会增加排队和重试压力。此时应明确返回 503 Service Unavailable，并让客户端按退避策略重试。

10.5 错误日志和错误响应不能混为一谈

对客户端：

{
  "error": {
    "code": "INTERNAL_ERROR",
    "message": "服务暂时不可用",
    "request_id": "req_..."
  }
}

对服务端日志：

request_id=req_...
exception=psycopg.errors.UniqueViolation
sqlstate=23505
route=POST /v1/orders
user_id=...
stacktrace=...

客户端需要稳定、有限、可处理的信息；运维人员需要足够的诊断上下文。两者的详细程度不应相同。

十一、测试 API 契约，而不是只测试函数

接口测试至少应覆盖以下行为：

输入合法 → 状态码和响应模型正确
输入非法 → 错误代码和字段位置正确
资源不存在 → 404
重复幂等请求 → 不创建第二条记录
相同键不同请求体 → 409
游标错误 → 400
最后一页 → next_cursor 为 null
超过限流 → 429 和 Retry-After 存在
旧版本 → 行为不受新版本破坏

幂等测试尤其要包含并发：

import asyncio


async def test_same_idempotency_key():
    results = await asyncio.gather(
        create_order_with_key("same-key"),
        create_order_with_key("same-key"),
    )

    assert results[0].id == results[1].id

单元测试只能证明某个函数在一次调用中的行为；幂等和限流的核心风险发生在并发状态变化中，必须使用集成测试或真实共享存储测试。

十二、最终模型：把六个主题连起来

可以把一次可靠的写请求抽象成：

请求进入
  ↓
版本路由选择契约
  ↓
认证和限流
  ↓
输入校验
  ↓
幂等键检查
  ↓
业务事务
  ↓
保存资源和幂等结果
  ↓
响应模型过滤
  ↓
统一错误和请求 ID

其中每一步都解决不同的问题：

契约解决“双方如何理解请求和响应”；
错误解决“失败后客户端如何行动”；
分页解决“有限响应如何访问无限或大型数据集”；
幂等解决“重试是否会造成重复副作用”；
限流解决“请求速率是否超过系统承受范围”；
版本解决“契约变化如何保持可演进”。

一个 API 的质量，不在于路由函数是否短，而在于这些边界是否明确：

正常请求有稳定输出
错误请求有可区分反馈
重复请求有确定结果
大量数据有稳定遍历
突发流量有可预测拒绝
契约变化有迁移路径

这才是 Python Web API 从“能够响应请求”走向“可以长期维护”的分界线。

系列导航与关联阅读

官方资料

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。