协程与异步 I/O 的边界、FastAPI 请求校验、路由分层和生产服务注意事项。

Python 异步与 FastAPI

学习目标：区分异步 I/O 与 CPU 计算，并用清晰的请求边界写出一个小型 API。

1. `async` 解决什么问题

协程（coroutine）在 await 等待网络或数据库 I/O 时让出执行权，使同一线程可以处理其他请求。它不会让 CPU 密集计算自动变快。

任务	适合的方式
异步 HTTP/数据库驱动	`async def` + `await`
同步库调用	同步端点或工作线程
大量 CPU 计算	进程池或独立任务服务

易错点：在 `async def` 内调用 `time.sleep()` 或同步网络客户端会阻塞事件循环；改用 `await asyncio.sleep()` 或对应异步客户端。

2. 一个最小 FastAPI 端点

以下示例需要安装 FastAPI 与 Uvicorn；为突出请求边界，数据只放在内存中。

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI()
items: dict[int, str] = {}

class ItemIn(BaseModel):
    name: str = Field(min_length=1, max_length=100)

class ItemOut(BaseModel):
    id: int
    name: str

@app.post("/items/{item_id}", response_model=ItemOut, status_code=201)
def create_item(item_id: int, payload: ItemIn) -> ItemOut:
    if item_id in items:
        raise HTTPException(status_code=409, detail="item already exists")
    items[item_id] = payload.name
    return ItemOut(id=item_id, name=payload.name)

@app.get("/items/{item_id}", response_model=ItemOut)
def get_item(item_id: int) -> ItemOut:
    name = items.get(item_id)
    if name is None:
        raise HTTPException(status_code=404, detail="item not found")
    return ItemOut(id=item_id, name=name)

运行示例：uvicorn app:app --reload，其中 app.py 是文件名。--reload 只用于本地开发。内存字典在多进程或重启后不会共享或持久化，真正项目应接入数据库。

3. 请求边界与依赖注入

HTTP 层负责路径、输入校验、状态码和响应格式；服务层负责业务规则；仓储层负责持久化。依赖注入用于传递数据库会话、认证信息与配置，避免在业务函数内部创建全局连接。

关键检查：每个端点写清成功响应、无效输入、找不到资源、冲突、无权限和服务器故障的行为。不要把所有异常都转成 200。

4. 生产环境要补的边界

为数据库操作设置超时与事务边界；外部调用设置连接和读取超时。
不在日志中输出密码、令牌或完整个人信息。
长任务用后台队列或独立任务系统；进程内临时任务不保证重启后继续执行。
用 HTTP API 与安全补鉴权、幂等和版本策略。

自测

为什么 CPU 密集循环放在 async def 中仍会拖慢其他请求？它没有等待点，会长时间占住事件循环。
内存字典适合作为生产数据库吗？不适合；重启丢失且多进程不共享。

5. 弄清一个请求如何运行

请求到达后，ASGI 服务器把它交给应用；路由先解析路径、查询参数和请求体，再调用端点函数，最后把返回值序列化。async def 端点在 await 支持异步的 I/O 时让出执行权；普通 def 端点适合同步驱动，框架可在线程池中执行它。把同步数据库调用直接写进 async def 会阻塞事件循环，影响其他请求。

选择原则：先确定依赖库是否真的提供异步 API。await 只用于可等待对象；把普通函数前面加 await 不会使其变成非阻塞调用。数据库连接数、下游服务容量和 CPU 仍会限制吞吐量。

6. 让输入、业务规则和输出各有位置

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI()

class CreateNote(BaseModel):
    title: str = Field(min_length=1, max_length=120)

class NoteOut(BaseModel):
    id: int
    title: str

def get_store() -> dict[int, str]:
    return {1: "示例"}  # 仅演示依赖注入；真实项目应返回仓储对象

@app.get("/notes/{note_id}", response_model=NoteOut)
def read_note(note_id: int, store: dict[int, str] = Depends(get_store)) -> NoteOut:
    title = store.get(note_id)
    if title is None:
        raise HTTPException(status_code=404, detail="note not found")
    return NoteOut(id=note_id, title=title)

CreateNote 是输入契约，NoteOut 是输出契约；它们不一定等于数据库表模型。Depends 使测试能够替换依赖，不过上例每次返回新字典，因此不具备持久性。真实项目中，依赖通常管理“每请求数据库会话”，业务函数接收仓储接口，HTTP 层负责错误映射。

7. 取消、超时和后台任务

客户端断开、请求超时、服务关闭都可能导致取消。把超时设置在实际 I/O 边界：数据库查询、HTTP 客户端、队列操作；不要认为一个总超时就能停止所有同步阻塞代码。后台任务若需要可靠重试和断点恢复，应写入持久队列；进程内临时任务只适合可丢失的小工作。

asyncio.create_task() 创建的任务必须有人管理生命周期、异常与取消，不能在请求函数里随手启动后不管。练习时可先用 asyncio.sleep() 模拟 I/O，再比较把 time.sleep() 放进 async def 后多个并发请求的表现。

8. 一条接口的验收清单与参考

为 POST /notes 列出成功 201、空标题校验失败、重复提交、数据库失败和未授权五类响应。每类都写一个测试，核对状态码与响应体，而不是只检查函数返回值。

02_异步与FastAPI