FastAPI自学教程（34） - 使用yield的依赖项

1. 基础用法

from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session

app = FastAPI()

class DBSession:
    def close(self): ...

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

• 使用yield代替return实现资源生命周期管理
• 执行流程：

    graph TD
    A[请求开始] --> B[执行yield前代码]
    B --> C[注入数据库会话到路由函数]
    C --> D[生成响应]
    D --> E[执行finally代码块]

• 核心特性：
  • yield前代码在响应生成前执行
  • finally代码块在响应发送后执行
  • 支持异步(async)和同步函数

2. 数据库会话管理

@app.get("/items/{item_id}")
async def read_item(
    item_id: str, 
    db: Session = Depends(get_db)
):
    item = db.query(...)
    return item

• 典型应用场景：
  • 数据库连接建立/关闭
  • 文件句柄打开/关闭
  • 网络连接创建/销毁
• 异常处理机制：
  • try...finally确保资源必被释放
  • 支持通过except捕获特定异常

3. 多级依赖项嵌套

async def dependency_a():
    dep_a = init_a()
    try:
        yield dep_a
    finally:
        dep_a.close()

async def dependency_b(dep_a=Depends(dependency_a)):
    dep_b = init_b(dep_a)
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)

@app.get("/multi")
async def multi_dep(dep_b=Depends(dependency_b)):
    ...

• 执行顺序原则：
  1. 子依赖项先初始化
  2. 父依赖项后初始化
  3. 退出时反向顺序释放
• 资源释放验证：
  • 父依赖项可访问子依赖实例
  • 确保依赖树正确解耦

4. 高级异常处理

class OwnerError(Exception): ...

async def get_user():
    try:
        yield "user"
    except OwnerError as e:
        raise HTTPException(400, detail=str(e))

• 支持在依赖项中抛出HTTPException
• 特殊限制：
  • yield后的异常处理不会影响已提交响应
  • HTTPException需在yield前抛出

5. 生产实践建议

1. 资源复用：通过@lru_cache缓存高频只读资源
2. 事务管理：在依赖项中实现commit/rollback逻辑

   async def transactional_db():
       db = get_db()
       try:
           yield db
           db.commit()
       except Exception:
           db.rollback()

3. 监控集成：在finally块记录资源使用指标
4. 安全警告：避免在yield后执行耗时操作（影响响应时间）