依赖注入系统：FastAPI 的复用机制

核心理念

FastAPI 的依赖注入系统允许你声明函数需要什么，框架自动提供。这不仅让代码可复用，还能自动处理资源清理、权限检查、数据库连接等横切关注点。

依赖的本质

依赖就是一个可调用对象（函数或类），返回值会被注入到路径操作函数的参数中。

最简单的依赖

from fastapi import Depends

def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

机制：

1. FastAPI 看到 Depends(common_parameters)
2. 调用 common_parameters，传入请求参数
3. 将返回值注入到 commons 参数

类作为依赖

class CommonQueryParams:
    def __init__(self, q: str = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends()):
    # Depends() 不传参数时，使用类型注解作为依赖
    return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}

嵌套依赖

依赖可以依赖其他依赖，形成依赖树：

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

def get_current_user(token: str = Header(), db: Database = Depends(get_db)):
    user = db.get_user_by_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

def get_admin_user(current_user: User = Depends(get_current_user)):
    if current_user.role != "admin":
        raise HTTPException(status_code=403, detail="Not an admin")
    return current_user

@app.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    admin: User = Depends(get_admin_user)  # 自动调用整个依赖链
):
    return {"deleted": user_id}

依赖树：

get_admin_user
    ↓
get_current_user
    ↓
get_db

yield 依赖：资源清理

def get_db():
    db = SessionLocal()
    try:
        yield db  # 请求处理在这里执行
    finally:
        db.close()  # 请求结束后自动清理

执行顺序：

1. db = SessionLocal() - 创建数据库会话
2. yield db - 注入到路径操作
3. 路径操作执行
4. db.close() - 自动清理

依赖缓存

同一个请求中，相同的依赖只会执行一次：

def expensive_operation():
    print("执行昂贵操作")
    return {"data": "cached"}

@app.get("/test")
async def test(
    dep1: dict = Depends(expensive_operation),
    dep2: dict = Depends(expensive_operation)
):
    # "执行昂贵操作" 只会打印一次
    return {"dep1": dep1, "dep2": dep2}

全局依赖

app = FastAPI(dependencies=[Depends(verify_token)])

# 所有路径都会先执行 verify_token

依赖覆盖（测试）

def fake_get_db():
    return MockDatabase()

app.dependency_overrides[get_db] = fake_get_db

# 测试时使用 Mock 数据库

与其他框架对比

特性	FastAPI	Flask	Spring Boot
声明式依赖	✅	❌	✅ (@Autowired)
自动资源清理	✅ (yield)	❌	✅ (@PreDestroy)
嵌套依赖	✅	❌	✅
依赖覆盖	✅	❌	✅

返回第二章概览