ECC fastapi-patterns 技能详解：FastAPI 异步 API、依赖注入与生产就绪模式

本文基于 ECC（Everything Claude Code）仓库中 .kiro/skills/fastapi-patterns/SKILL.md 技能文档展开，系统讲解面向生产的 FastAPI 服务模式：应用工厂与 lifespan 管理、Pydantic v2 请求/响应模型拆分、依赖注入中的数据库会话事务语义、异步端点规范、集中式异常处理、OpenAPI 定制、基于 dependency_overrides 的测试方案，以及安全与性能两套生产检查清单。读完后你将掌握一套可直接复制到项目中的 FastAPI 分层架构范式，并了解 ECC 如何用配套的 fastapi-reviewer 代理与 /fastapi-review 命令将这套模式固化为自动化审查流程。

技能定位：什么场景下使用 fastapi-patterns

该技能文档（.kiro/skills/fastapi-patterns/SKILL.md）在 frontmatter 中声明了自己的用途：

name: fastapi-patterns
description: FastAPI patterns for async APIs, dependency injection, Pydantic request and response models, OpenAPI docs, tests, security, and production readiness.
origin: community

文档给出的适用场景（When to Use）覆盖了 FastAPI 项目从搭建到审查的完整生命周期：

• 构建或评审一个 FastAPI 应用；
• 划分 routers、schemas、dependencies 与数据库访问的职责边界；
• 编写需要调用数据库或外部服务的异步端点；
• 添加认证、授权、OpenAPI 文档、测试或部署配置；
• 检查一个 FastAPI PR 中是否存在可复制的坏示例和面向生产的风险。

这套技能在 ECC 仓库中有两个存放位置：.kiro/skills/ 下的是面向 Kiro IDE/CLI 的精简社区版（origin: community），而 skills/fastapi-patterns/SKILL.md 是内容更丰富的 ECC 原生版（origin: ECC），额外覆盖了 pydantic-settings 配置、事务性服务层、httpx/pytest 完整测试夹具与反模式清单。两版可以对照阅读，本文以 .kiro 版为骨架，在相关章节引用增强版内容作补充。

核心设计思想：把应用当作显式依赖之上的薄 HTTP 层

技能文档的 How It Works 一节给出了总体架构原则：将 FastAPI 应用视为一层薄薄的 HTTP 层，其下是显式的依赖与业务代码。具体职责划分为：

模块	职责
main.py	应用构建、中间件、异常处理器、路由注册
schemas/	Pydantic 请求与响应模型
dependencies.py	数据库、认证、分页等请求作用域依赖
services/ 或 crud/	业务与持久化操作
tests/	通过覆盖依赖来测试，而不是打开生产资源

文档还强调了两个落地要求：优先使用小颗粒度的 router 和显式的 response_model 声明；让原始 ORM 对象、密钥和框架全局变量远离响应 schema。这一思想与仓库中的规则文件 rules/python/fastapi.md 完全一致——该规则通过 frontmatter 中的路径匹配（**/app/**/*.py、**/fastapi/**/*.py、**/*_api.py）在编辑 FastAPI 相关 Python 文件时自动生效，重复了同一组约束：应用构建放在 create_app() 中、router 保持薄层、请求/更新/响应 schema 分离、数据库会话与认证放在依赖中。

项目布局

文档推荐的目录结构如下，每一层都对应上面"职责划分表"中的一行：

app/
|-- main.py
|-- config.py
|-- dependencies.py
|-- exceptions.py
|-- api/
|   `-- routes/
|       |-- users.py
|       `-- health.py
|-- core/
|   |-- security.py
|   `-- middleware.py
|-- db/
|   |-- session.py
|   `-- crud.py
|-- models/
|-- schemas/
`-- tests/

几个值得注意的取舍：

• config.py 独立存放配置（增强版 skills/fastapi-patterns/SKILL.md 中使用 pydantic-settings 的 BaseSettings 从 .env 读取，database_url、secret_key 等敏感项必须来自环境变量）；
• exceptions.py 单独成文件，承载统一异常注册逻辑；
• db/ 目录把会话工厂（session.py）与 CRUD 操作（crud.py）分开，避免在路由层直接写持久化代码；
• core/ 放安全与中间件实现，与 dependencies.py 中消费这些实现的 FastAPI 依赖解耦。

应用工厂：create_app 与 lifespan

文档要求使用应用工厂模式（Application Factory），目的是让测试和 worker 进程都能以受控配置构建应用实例：

from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.api.routes import health, users
from app.config import settings
from app.db.session import close_db, init_db
from app.exceptions import register_exception_handlers


@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()
    yield
    await close_db()


def create_app() -> FastAPI:
    app = FastAPI(
        title=settings.api_title,
        version=settings.api_version,
        lifespan=lifespan,
    )

    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.cors_origins,
        allow_credentials=bool(settings.cors_origins),
        allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
        allow_headers=["Authorization", "Content-Type"],
    )

    register_exception_handlers(app)
    app.include_router(health.router, prefix="/health", tags=["health"])
    app.include_router(users.router, prefix="/api/v1/users", tags=["users"])
    return app


app = create_app()

这段代码中有三个值得细读的设计点：

1. lifespan 替代 on_event：@asynccontextmanager 装饰的 lifespan 在应用启动时 await init_db()、在关闭时 await close_db()，用单一异步上下文管理器管理"启动—运行—关闭"全生命周期，取代了已废弃的 @app.on_event("startup") 写法。从增强版技能可以看到另一种常见实现：lifespan 内用 async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) 建表、关闭时 await engine.dispose() 释放连接池，并注释说明生产环境应改用 Alembic 管理 schema。
2. CORS 配置的安全写法：allow_credentials=bool(settings.cors_origins) 把"是否允许携带凭证"与"是否配置了来源列表"绑定——没有显式来源时自动关闭凭证模式。
3. 路由分层注册：健康检查走 /health（不带 API 版本前缀，便于负载均衡器探针），业务路由走 /api/v1/users，版本号固化在 URL 前缀里。

文档同时给出了一条硬性警告，这也是 fastapi-reviewer 代理将其列为 High 级问题的原因：不要使用 allow_origins=["*"] 搭配 allow_credentials=True——浏览器会拒绝这种组合，Starlette 对携带凭证的请求也不允许通配来源。CORS 来源应当按环境区分（见下文安全检查清单）。

Pydantic 模式：请求、更新、响应模型分离

文档要求"把请求、更新、响应模型分开"，并给出了完整的四模型拆分示例：

from datetime import datetime
from typing import Annotated
from uuid import UUID

from pydantic import BaseModel, ConfigDict, EmailStr, Field


class UserBase(BaseModel):
    email: EmailStr
    full_name: Annotated[str, Field(min_length=1, max_length=100)]


class UserCreate(UserBase):
    password: Annotated[str, Field(min_length=12, max_length=128)]


class UserUpdate(BaseModel):
    email: EmailStr | None = None
    full_name: Annotated[str | None, Field(min_length=1, max_length=100)] = None


class UserResponse(UserBase):
    model_config = ConfigDict(from_attributes=True)

    id: UUID
    created_at: datetime
    updated_at: updated_at: datetime

（上例 updated_at 一行按原文为 updated_at: datetime，此处保留原文排版。）

这四个类各承担一种职责，继承关系刻意"短平快"：

• UserBase：创建与响应共享的字段及约束（EmailStr 类型校验邮箱格式，Field(min_length/max_length) 表达长度规则），用 Annotated 把类型与约束合并在一个注解里；
• UserCreate：在 Base 之上追加 password，且密码约束为 12～128 位——这是比常见示例（8 位起）更严格的策略；
• UserUpdate：所有字段默认为 None，配合 PATCH 语义实现"只更新客户端显式传入的字段"。增强版技能在服务层用 payload.model_dump(exclude_unset=True) 消费这种模式，确保未传入的字段不会被覆盖为 None；
• UserResponse：ConfigDict(from_attributes=True) 允许直接从 ORM 对象（如 SQLAlchemy 模型）按属性名构造响应，避免手写逐字段映射；同时用 UUID 作为主键类型、补充 created_at/updated_at 审计字段。

文档在此节末尾给出一条安全红线：响应模型绝不允许包含密码哈希、访问令牌、刷新令牌或内部授权状态。这一点在仓库的审查体系中被反复强化——agents/fastapi-reviewer.md 把"密码、令牌哈希或内部认证字段暴露于响应模型"列为 Critical 级发现；rules/python/fastapi.md 也在 Schemas 一节重复了同样的禁令。

依赖注入：数据库会话的事务语义与当前用户解析

文档要求用 FastAPI 依赖注入管理请求作用域资源，核心是两个依赖：

from collections.abc import AsyncIterator
from uuid import UUID

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.security import decode_token
from app.db.session import session_factory
from app.models.user import User


oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")


async def get_db() -> AsyncIterator[AsyncSession]:
    async with session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise


async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
) -> User:
    payload = decode_token(token)
    user_id = UUID(payload["sub"])
    user = await db.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
    return user

get_db 的设计值得展开：它是一个生成器依赖，async with session_factory() 保证会话一定来自连接池且一定被关闭；yield 之后的 commit() 意味着只要请求正常处理完毕就提交，任何异常先回滚再抛出——事务边界被收敛到了整个请求，路由处理函数自身不需要关心 commit/rollback。需要注意一个隐含语义：由于提交发生在 yield 之后，依赖注入的依赖链（get_current_user 依赖 get_db）中内层依赖先建立、后销毁，会话的生命周期严格覆盖所有路由逻辑。

get_current_user 展示了 JWT 鉴权的最小闭环：OAuth2PasswordBearer 从 Authorization: Bearer <token> 头提取令牌，decode_token 解码后取 sub 声明还原用户 ID，UUID(payload["sub"]) 做类型转换，查库失败即返回 401。文档明确要求 JWT 解码必须校验 issuer、audience、过期时间与签名算法（见安全清单），而 fastapi-reviewer 代理把"认证依赖可被绕过或不校验过期/签名"同样列为 Critical。

该节末尾的告诫：不要在路由处理器内部联行创建 session、client 或凭据对象。rules/python/fastapi.md 用同样的话强化了这一点——不要 SessionLocal() 出现在 handler 里。增强版技能还给出了配套的代码风格建议：用类型别名收敛重复的 Depends 声明，如 DbDep = Annotated[AsyncSession, Depends(get_db)]，让路由签名保持整洁。

异步端点：I/O 走 async，外部调用用 AsyncClient

文档给出的列表端点示例：

from fastapi import APIRouter, Depends, Query
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.dependencies import get_current_user, get_db
from app.models.user import User
from app.schemas.user import UserResponse


router = APIRouter()


@router.get("/", response_model=list[UserResponse])
async def list_users(
    limit: int = Query(default=50, ge=1, le=100),
    offset: int = Query(default=0, ge=0),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    result = await db.execute(
        select(User).order_by(User.created_at.desc()).limit(limit).offset(offset)
    )
    return result.scalars().all()

这个端点浓缩了文档的几条规范：

• 端点保持 async def 并使用 async 库：数据库操作用 await db.execute(select(...))（async SQLAlchemy 执行），而不是同步 Session 的 db.query(...)。文档明确规定：从 async handler 调用外部 HTTP 时使用 httpx.AsyncClient，不要在异步路由里调用 requests——阻塞调用会卡住事件循环，fastapi-reviewer 把"async 路由中的阻塞数据库或 HTTP 客户端"列为 High 级问题；
• 查询参数用 Query 约束：limit 默认 50、ge=1、le=100，offset 默认 0、ge=0——分页上限在 API 层强制，防止客户端用超大 limit 打爆数据库；
• 显式 response_model=list[UserResponse]：输出经过 Pydantic 序列化，OpenAPI 文档自动获得准确的响应 schema，同时杜绝了意外泄漏 ORM 内部字段；
• 依赖注入而非手工取会话：db 和 current_user 都来自 Depends，路由函数本身是纯函数式的。

增强版技能在此之上补充了两条反模式对照：坏示例是"业务逻辑写在路由处理器里"（handler 中直接 bcrypt.hash + db.add + db.commit），好示例是"薄路由 + 事务性 service 方法"；坏示例是"async 路由中调用同步 SQLAlchemy"，好示例是"用 await db.execute(select(Item))"。

集中式异常处理：稳定错误响应形状

文档要求把领域异常集中定义、保持响应形状稳定：

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


class ApiError(Exception):
    def __init__(self, status_code: int, code: str, message: str):
        self.status_code = status_code
        self.code = code
        self.message = message


def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(ApiError)
    async def api_error_handler(request: Request, exc: ApiError):
        return JSONResponse(
            status_code=exc.status_code,
            content={"error": {"code": exc.code, "message": exc.message}},
        )

要点在于：业务层（services/crud）抛出携带 status_code、机器可读 code 与人类可读 message 的 ApiError，由 register_exception_handlers 在应用工厂中统一注册处理器，转换为固定的 {"error": {"code": ..., "message": ...}} JSON 形状。客户端因此可以依赖稳定的错误信封做解析，而不用逐个端点约定错误格式。ApiError 与 FastAPI 内建的 HTTPException 形成分工：后者处理框架级问题（401/404/422），前者承载领域语义（如"邮箱已注册"）。增强版技能给出了这种分工的实际用法：service 层抛 DuplicateUserError，路由层捕获后转成 HTTPException(400, "Email already registered")，并且注释强调唯一性冲突应依赖数据库约束的原子性（捕获 IntegrityError）而非应用层"先查后插"的竞态式预检。

OpenAPI 定制：正确改写 app.openapi

文档特别强调一个易错点：把自定义 OpenAPI 可调用对象赋值给 app.openapi，而不是只调用一次该函数：

from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi


def install_openapi(app: FastAPI) -> None:
    def custom_openapi():
        if app.openapi_schema:
            return app.openapi_schema
        app.openapi_schema = get_openapi(
            title="Service API",
            version="1.0.0",
            routes=app.routes,
        )
        return app.openapi_schema

    app.openapi = custom_openapi

app.openapi 是 FastAPI 生成 /openapi.json 时调用的方法，直接替换它才能注入自定义逻辑（追加安全方案、改写组件、补充服务端点列表等）。函数内先检查 app.openapi_schema 是否已缓存，是标准的防重复计算写法。fastapi-reviewer 的审查范围中专门包含"OpenAPI 元数据与已文档化的响应模型"，Medium 级问题里就有"OpenAPI 文档缺少响应模型或错误响应描述"。

测试：覆盖依赖而不是内部助手

文档对测试方案的第一原则是：覆盖路由实际使用的 Depends 依赖，而不是路由处理器根本不引用的内部助手。示例夹具：

import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession

from app.dependencies import get_db
from app.main import create_app


@pytest.fixture
async def client(test_session: AsyncSession):
    app = create_app()

    async def override_get_db():
        yield test_session

    app.dependency_overrides[get_db] = override_get_db
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as test_client:
        yield test_client
    app.dependency_overrides.clear()

这条链路值得逐步理解：

1. create_app() 复用应用工厂构建测试应用，配置可注入测试 settings（这正是工厂模式的价值——如果 main.py 是模块级硬编码的 app，测试就无法替换配置）；
2. app.dependency_overrides[get_db] = override_get_db 让所有 Depends(get_db) 的路由拿到夹具提供的 test_session，生产数据库连接池、真实认证链路全部不经过；注意覆盖的键必须是路由签名中那个依赖函数对象本身，fastapi-reviewer 把"测试覆盖打错了依赖"列为 High 级问题，rules/python/fastapi.md 则要求测试结束清空 app.dependency_overrides——夹具末尾的 app.dependency_overrides.clear() 正是对此的落实；
3. AsyncClient(transport=ASGITransport(app=app)) 直接在进程内驱动 ASGI 应用，无真实网络端口，异步应用用异步测试客户端（规则文件同样要求"async 应用优先异步测试客户端"）。

增强版技能给出了完整的 conftest.py 参考实现：用 sqlite+aiosqlite:///:memory: 内存库、pytest_asyncio 夹具在 setup 阶段 Base.metadata.create_all / teardown 阶段 drop_all，并串联出 registered_user → auth_token → auth_client 三级夹具，覆盖"注册 → 登录取令牌 → 带 Bearer 头请求"的完整鉴权测试路径，可直接作为编写集成测试的模板。

生产安全检查清单

文档给出八条安全要求，每一条都能在仓库的审查规则中找到对应物：

检查项	说明	审查体系中的对应
密码哈希	使用 argon2-cffi、bcrypt 或当前 passlib 兼容的哈希器	增强版技能示例用 CryptContext(schemes=["bcrypt"])
JWT 校验	校验 issuer、audience、过期时间与签名算法	fastapi-reviewer Critical：认证依赖不校验过期/签名
CORS	来源按环境区分	High：allow_origins=["*"] + 凭证
限流	对认证端点与写密集端点加限流	rules/python/fastapi.md Security 节同样要求
请求体校验	所有请求体使用 Pydantic 模型	审查范围"写入端点缺少请求校验"（High）
SQL 注入	使用 ORM 参数绑定或 SQLAlchemy Core 表达式，绝不用 f-string 拼 SQL	Critical：字符串插值构造 SQL
日志脱敏	令牌、Authorization 头、Cookie、密码不得入日志	规则文件要求日志中脱敏凭据与 Cookie
依赖审计	CI 中运行依赖审计工具	仓库 CI 脚本目录（scripts/ci/）承担同类质量门禁角色

这份清单的价值在于把"安全"从模糊概念变成可勾选的工程动作：前四条防的是认证与传输层的常见失守，后四条防的是数据层与供应链层。

性能检查清单

文档同时给出六条性能要求：

• 显式配置数据库连接池（不依赖驱动默认值）；
• 列表端点一律加分页（上文 Query(le=100) 的 limit 上限即具体实现）;
• 警惕 N+1 查询，有意识地使用 eager loading（selectinload/joinedload），而不是默认懒加载；
• async 路径使用 async HTTP/数据库客户端（与端点一节呼应）；
• 加压缩（gzip 等）之前先评估 payload 大小与 CPU 开销的权衡；
• 昂贵但稳定的读操作放在显式失效策略的缓存之后——"explicit invalidation"意味着缓存必须与写路径联动，而非无限期缓存。

fastapi-reviewer 的 Medium 级问题里对应包含"列表端点缺少分页""外部 HTTP 客户端缺少超时设置"，可以看到这份检查清单与审查代理是同一套标准的两种表达。

配套体系：fastapi-reviewer 代理与 /fastapi-review 命令

技能文档末尾的 See Also 指向了 ECC 的三层配套：fastapi-reviewer 代理、/fastapi-review 命令、以及 python-patterns、python-testing、api-design 三个相邻技能（仓库根目录对应 skills/python-patterns/SKILL.md、skills/python-testing/SKILL.md、skills/api-design/SKILL.md，.kiro/skills/ 下亦有同名精简版）。

commands/fastapi-review.md 定义了调用方式 /fastapi-review [file-or-directory]，其审查面与本技能的模式面几乎一一对应：应用工厂/路由边界/中间件/异常处理器、Pydantic 请求与响应分离、数据库会话/认证/分页/配置的依赖注入、async 数据库与外部 HTTP 模式、CORS/认证/限流/日志/密钥处理、OpenAPI 元数据与已文档化的响应模型、测试客户端与依赖覆盖。期望输出格式固定为 [SEVERITY] 问题标题 + File: 文件:行号 + Issue + Fix 四段，保证发现可执行、可定位。

agents/fastapi-reviewer.md 则定义了这个"资深 FastAPI 审查者"的完整工作流：先用 Read/Grep/Glob 定位 main.py/app.py 入口，识别 routers/schemas/dependencies/会话配置/测试；安全时运行本地检查（pytest、ruff、mypy、uv run pytest）；先审变更文件、再看相邻定义以坐实发现。其发现分级把本技能的各条红线落实为可判定的严重度：

• Critical：硬编码密钥/令牌；字符串插值拼 SQL；响应模型暴露密码、令牌哈希或内部认证字段；认证依赖可被绕过或不校验过期/签名；
• High：async 路由中的阻塞数据库/HTTP 客户端；handler 内联创建数据库会话；测试覆盖打错依赖；allow_origins=["*"] + 凭证 CORS；写入端点缺少请求校验;
• Medium：列表端点缺分页；OpenAPI 缺响应模型/错误描述；应下沉到 service/dependency 的重复路由逻辑；外部 HTTP 客户端缺超时。

代理还约束了边界（Out of Scope）：不审查非 FastAPI 框架（除非它们直接与 FastAPI 应用交互）、不做已由 python-reviewer 覆盖的通用 Python 风格审查、不为没有具体问题的依赖添加而辩护。最后要求输出 Tests checked:（跑了什么命令或为何跳过）与 Residual risk:（无法验证的重要项），把审查结论的置信度显式化。

这套"技能（模式知识）→ 规则（文件级自动约束）→ 命令（一键触发）→ 代理（执行审查）"的分工，正是 ECC 作为 agent harness 的核心思路：同一套 FastAPI 最佳实践既指导 Agent 写代码，也作为审查 Agent 的判分标准。

在 Kiro 中安装与使用这份技能

.kiro 目录是 ECC 面向 Kiro IDE/CLI 的集成层（见 .kiro/README.md）。技能安装后在 Kiro 会话中输入 / 打开技能菜单，选择 fastapi-patterns 即可让 Agent 按本文明确的工作流与检查清单引导开发；fastapi-reviewer 则可以通过 /fastapi-reviewer 显式调用（IDE 中）或 /agent swap 切换（CLI 中）。

安装通过 .kiro/install.sh 完成，脚本行为从源码可以直接确认：

cd .kiro
./install.sh /path/to/your/project   # 安装到指定项目
./install.sh                        # 安装到当前目录
./install.sh ~                      # 全局安装到 ~/.kiro/

脚本按 agents skills steering hooks scripts settings 六个子目录做非破坏性拷贝——从源码可见每个文件都先判断 [ ! -f "$TARGET/.kiro/..." ] 才复制（技能则按目录粒度判断 [ ! -d ... ]），已存在的文件一律跳过不覆盖，因此安装不会破坏你已有的定制，重复执行也是安全的。

小结

.kiro/skills/fastapi-patterns/SKILL.md 提供的是一套自洽的 FastAPI 生产模式：以应用工厂管理生命周期与中间件，用 Base/Create/Update/Response 四类 Pydantic 模型锁定输入输出边界，用请求作用域依赖收敛数据库事务与鉴权，用 async 客户端保持事件循环不阻塞，用集中式异常与 app.openapi 赋值保持契约稳定，用 dependency_overrides 让测试完全脱离生产资源，最后用安全与性能两套清单做上线前核对。配合仓库中 fastapi-reviewer 代理、/fastapi-review 命令与路径触发的 rules/python/fastapi.md 规则，这套模式从"文档建议"变成了"可自动执行的审查标准"——这也是 ECC 技能体系的典型组织方式：模式、规则、命令、代理四者同源同标准。