Fastapi async api

何时使用

适用：

• 用 FastAPI（0.100+，建议 Annotated 依赖注入）搭建高并发异步 REST/WebSocket API 或微服务。
• 需要 async SQLAlchemy 2.0（asyncpg/aiomysql）、Pydantic V2 校验与序列化、Alembic 迁移。
• 落地 OAuth2 + JWT 鉴权、RBAC、限流、连接池、Redis 缓存、健康检查与可观测性。
• 排查异步端点性能问题（N+1、阻塞调用、连接耗尽）。

不该用：

• 同步框架场景（Flask/Django 同步视图、WSGI），异步收益不成立。
• 纯前端、静态站点或非 Python 服务。
• 仅需简单脚本而非长期维护的服务 API。

边界：本技能给出工程范式与可运行骨架，不替代针对你环境的压测、安全审计与代码评审。缺少关键输入（数据库类型、鉴权要求、部署目标）时先澄清再动手。

步骤

1. 厘清需求：识别可异步化的 IO 点（DB、外部 HTTP、消息队列），确定鉴权与并发目标。
2. 先定契约：用 Pydantic V2 模型设计 request/response schema，再写端点。
3. 实现端点：async def + 依赖注入（DB session、当前用户），统一异常处理。
4. 接数据层：async SQLAlchemy 2.0 引擎 + 连接池，避免在 async 路径调用同步阻塞库。
5. 加鉴权与防护：OAuth2 + JWT，CORS、限流、输入校验。
6. 写异步测试：pytest + pytest-asyncio + httpx.AsyncClient / TestClient 覆盖边界。
7. 优化：缓存（Redis）、eager loading 防 N+1、分页（推荐游标分页）、响应压缩。
8. 可观测与部署：结构化日志、健康检查、Uvicorn/Gunicorn 生产配置、Docker 多阶段构建。

指令

• async 优先：IO 端点用 async def；CPU 密集或仅同步库的逻辑放线程池（run_in_threadpool / asyncio.to_thread），绝不在事件循环里直接阻塞。
• 类型安全：全程类型注解，依赖用 Annotated[T, Depends(...)]。
• 启停资源用 lifespan：在 lifespan 中建/销引擎、连接池、缓存客户端，避免每请求重建。
• 错误处理：自定义 exception handler，返回结构化错误体，不向客户端泄漏内部异常。
• 配置用 Pydantic Settings + 环境变量，遵循 12-factor。
• 需要详细范式时再展开数据层、鉴权、部署各环节，按需取用。

示例

最小异步骨架（FastAPI + async SQLAlchemy 2.0 + Pydantic V2 + lifespan）：

from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy.ext.asyncio import (
    create_async_engine, async_sessionmaker, AsyncSession,
)
from sqlalchemy import select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

engine = create_async_engine(
    "postgresql+asyncpg://user:pwd@localhost/app",
    pool_size=20, max_overflow=10, pool_pre_ping=True,
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase): ...

class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]

class ItemOut(BaseModel):
    id: int
    name: str
    model_config = {"from_attributes": True}  # Pydantic V2

@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await engine.dispose()

app = FastAPI(lifespan=lifespan)

async def get_db():
    async with SessionLocal() as session:
        yield session

@app.get("/items/{item_id}", response_model=ItemOut)
async def read_item(item_id: int, db: Annotated[AsyncSession, Depends(get_db)]):
    item = await db.get(Item, item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

异步测试（httpx + pytest-asyncio）：

import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest.mark.asyncio
async def test_read_item():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        resp = await ac.get("/items/1")
    assert resp.status_code in (200, 404)

生产启动（多 worker）：

gunicorn app.main:app -k uvicorn.workers.UvicornWorker -w 4 --bind 0.0.0.0:8000
# 开发热重载
uvicorn app.main:app --reload

注意事项

• N+1 查询：用 selectinload / joinedload 显式 eager loading；别在循环里逐条 await 查询。
• 连接耗尽：合理设置 pool_size/max_overflow，session 用完即关（依赖 yield 自动回收）。
• 别混用同步阻塞库（如同步 requests、同步 ORM 调用）到 async 路径，会拖垮整个事件循环；HTTP 调用用 httpx.AsyncClient。
• Pydantic V2 与 V1 API 不兼容：用 model_config、from_attributes、model_dump()，勿照搬 V1 的 Config/orm_mode。
• JWT：校验签名、过期与受众；刷新令牌与访问令牌分离，密钥从环境注入。
• 分页大数据集优先游标分页，避免深 offset 全表扫描。

互见

• 数据库迁移：Alembic（async 模板）配套使用。
• 鉴权深化：OAuth2 / JWT / RBAC 专项实践。
• 部署：Docker 多阶段构建、Kubernetes/Helm 与 CI/CD。

---

采编自 sickn33/antigravity-awesome-skills（MIT）。