fastapi-best-practices

适用场景

在编写 FastAPI 代码时使用此技能。它介绍当前的最佳实践，并帮助避免 AI Agent 使用过时模式时常犯的错误。

关键规则

1. I/O 密集型端点使用 async def，CPU 密集型使用 def

错误（Agent 常这样做）：

@app.get("/users")
def get_users():
    users = db.query(User).all()
    return users

@app.get("/data")
async def get_data():
    result = heavy_computation()
    return result

正确：

@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    return result.scalars().all()

@app.get("/data")
def get_data():
    return heavy_computation()

**原因：**FastAPI 在事件循环中运行异步端点；同步端点则在线程池中运行。对于 I/O 操作（数据库、HTTP、文件），使用 async 以避免阻塞。对于 CPU 密集型工作，使用 def；将其设为异步会阻塞事件循环。

2. 使用 Depends() 进行依赖注入

错误（Agent 常这样做）：

db = get_database()

@app.get("/items")
async def get_items():
    return db.query(Item).all()

正确：

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/items")
async def get_items(db: Annotated[Session, Depends(get_db)]):
    return db.query(Item).all()

**原因：**全局数据库连接会泄漏、难以测试，还会绕过 FastAPI 的依赖系统。Depends() 可提供适当的作用域和清理机制，并支持测试覆盖。

3. 使用 Pydantic v2 模式

错误（Agent 常这样做）：

from pydantic import validator

class Item(BaseModel):
    name: str
    price: float

    class Config:
        orm_mode = True

    @validator("price")
    def price_positive(cls, v):
        if v <= 0:
            raise ValueError("must be positive")
        return v

正确：

from pydantic import BaseModel, field_validator, ConfigDict

class Item(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    name: str
    price: float

    @field_validator("price")
    @classmethod
    def price_positive(cls, v: float) -> float:
        if v <= 0:
            raise ValueError("must be positive")
        return v

**原因：**Pydantic v1 的 validator、Config 和 orm_mode 已弃用。请使用 field_validator、model_validator、ConfigDict 和 from_attributes。

4. 使用 lifespan 上下文管理器

错误（Agent 常这样做）：

@app.on_event("startup")
async def startup():
    app.state.db = await create_pool()

@app.on_event("shutdown")
async def shutdown():
    await app.state.db.close()

正确：

from contextlib import asynccontextmanager

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

app = FastAPI(lifespan=lifespan)

**原因：**on_event 已弃用。lifespan 上下文管理器为启动和关闭提供统一位置，并能正确安排资源的处理顺序。

5. 使用 BackgroundTasks 执行即发即弃的工作

错误（Agent 常这样做）：

@app.post("/send-email")
async def send_email(email: str):
    asyncio.create_task(send_email_async(email))
    return {"status": "queued"}

正确：

@app.post("/send-email")
async def send_email(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email_async, email)
    return {"status": "queued"}

**原因：**asyncio.create_task 可能在请求结束后仍继续运行，并且在关闭时不会等待其完成。BackgroundTasks 会在发送响应后运行，并与请求生命周期绑定。

6. 使用 APIRouter 组织路由

错误（Agent 常这样做）：

# main.py - 500 lines of routes
@app.get("/users")
@app.get("/users/{id}")
@app.post("/items")
@app.get("/items")

正确：

# main.py
app.include_router(users.router, prefix="/users", tags=["users"])
app.include_router(items.router, prefix="/items", tags=["items"])

# routers/users.py
router = APIRouter()
@router.get("/")
@router.get("/{id}")

**原因：**单文件应用难以维护。APIRouter 支持采用 routers/、models/、services/、dependencies/ 结构。

7. 使用 response_model 验证输出

错误（Agent 常这样做）：

@app.get("/items/{id}")
async def get_item(id: int):
    item = await db.get(Item, id)
    return {"id": item.id, "name": item.name}

正确：

@app.get("/items/{id}", response_model=ItemOut)
async def get_item(id: int, db: Session = Depends(get_db)):
    item = await db.get(Item, id)
    if not item:
        raise HTTPException(status_code=404)
    return item

**原因：**原始 dict 会绕过验证和 OpenAPI。response_model 可确保架构一致性、序列化和文档准确性。

8. 使用 fastapi.status 中的状态码

错误（Agent 常这样做）：

raise HTTPException(status_code=404, detail="Not found")
raise HTTPException(status_code=401, detail="Unauthorized")

正确：

from fastapi import status

raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized")

**原因：**魔法数字容易出错。状态常量能自我说明，并符合 HTTP 规范。

9. 对依赖使用 Annotated

错误（Agent 常这样做）：

@app.get("/me")
async def read_me(current_user: User = Depends(get_current_user)):
    return current_user

正确：

@app.get("/me")
async def read_me(current_user: Annotated[User, Depends(get_current_user)]):
    return current_user

**原因：**Annotated 是推荐的 FastAPI 模式。它将类型和依赖放在一处，并支持复用依赖。

10. 使用 pydantic-settings 进行配置

错误（Agent 常这样做）：

import os
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///db.sqlite")

正确：

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str = "sqlite:///db.sqlite"
    debug: bool = False
    model_config = {"env_file": ".env"}

settings = Settings()

**原因：**os.getenv 不提供验证或类型信息。BaseSettings 支持验证、加载 .env 文件和类型安全。

模式

• 在 routers/ 中定义路由器，并设置前缀和标签
• 将共享依赖放在 dependencies.py 中
• 错误处理使用带状态常量的 HTTPException
• 使用经过验证的 Path、Query、Body、Header（min_length、ge、le）
• 使用 app.add_exception_handler 注册自定义异常处理器
• 谨慎使用中间件；顺序很重要（最先添加的中间件在请求处理中最后运行）

反模式

• 不要使用 @app.on_event("startup") 或 @app.on_event("shutdown")
• 不要使用 asyncio.create_task 执行作用于请求范围的后台工作
• 不要将数据库、缓存或配置存储在全局变量中
• 不要使用 Pydantic v1 的 @validator 或 class Config
• 不要在没有 response_model 的情况下返回原始 dict
• 不要使用魔法数字表示状态码
• 不要将所有路由都放在 main.py 中