依赖注入系统

依赖注入的核心概念

依赖注入（Dependency Injection，DI）是一种设计模式：组件不自己创建所需依赖，而是声明需要什么，由框架负责创建并注入。FastAPI 的 Depends() 实现了这一模式，带来代码复用、逻辑解耦和易于测试等优势。

函数依赖

from fastapi import FastAPI, Depends, HTTPException
from typing import Annotated

app = FastAPI()

# ── 通用分页参数依赖 ──────────────────────────────────────
async def pagination(
    page: int = 1,
    size: int = 20
) -> dict:
    if size > 100:
        raise HTTPException(status_code=400, detail="每页最多 100 条")
    skip = (page - 1) * size
    return {"skip": skip, "limit": size, "page": page}

# Annotated 写法（推荐，类型提示更清晰）
PaginationDep = Annotated[dict, Depends(pagination)]

@app.get("/users/")
async def list_users(pg: PaginationDep):
    # pg = {"skip": 0, "limit": 20, "page": 1}
    return {"page": pg["page"], "users": []}

@app.get("/items/")
async def list_items(pg: PaginationDep):
    return {"page": pg["page"], "items": []}

类依赖

当依赖需要持有状态（如配置参数）或更复杂的初始化逻辑时，使用类作为依赖：

from fastapi import FastAPI, Depends
from typing import Annotated

app = FastAPI()

# ── 类依赖：有状态的过滤器 ────────────────────────────────
class ItemFilter:
    def __init__(
        self,
        category: str | None = None,
        min_price: float = 0,
        max_price: float = float("inf"),
        in_stock_only: bool = False
    ):
        self.category = category
        self.min_price = min_price
        self.max_price = max_price
        self.in_stock_only = in_stock_only

    def apply(self, items: list) -> list:
        result = items
        if self.category:
            result = [i for i in result if i["category"] == self.category]
        result = [i for i in result
                  if self.min_price <= i["price"] <= self.max_price]
        if self.in_stock_only:
            result = [i for i in result if i["stock"] > 0]
        return result

# 类本身就是可调用的，FastAPI 会实例化它并注入
FilterDep = Annotated[ItemFilter, Depends(ItemFilter)]

@app.get("/catalog/")
async def get_catalog(filters: FilterDep):
    # GET /catalog/?category=电子&min_price=100&in_stock_only=true
    all_items = [
        {"id": 1, "name": "笔记本", "category": "电子", "price": 5999, "stock": 10},
        {"id": 2, "name": "键盘", "category": "电子", "price": 299, "stock": 0},
    ]
    return {"items": filters.apply(all_items)}

依赖链（嵌套 Depends）

from fastapi import FastAPI, Depends, HTTPException, Header
from typing import Annotated

app = FastAPI()

# ── 第一层：提取 Token ────────────────────────────────────
async def get_token(authorization: Annotated[str | None, Header()] = None) -> str:
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="缺少认证 Token")
    return authorization.removeprefix("Bearer ")

# ── 第二层：验证 Token，获取用户（依赖第一层）────────────
async def get_current_user(token: Annotated[str, Depends(get_token)]) -> dict:
    # 实际场景：解码 JWT，查询数据库
    if token != "valid-token":
        raise HTTPException(status_code=401, detail="Token 无效或已过期")
    return {"id": 1, "username": "alice", "role": "user"}

# ── 第三层：要求管理员权限（依赖第二层）──────────────────
async def require_admin(
    current_user: Annotated[dict, Depends(get_current_user)]
) -> dict:
    if current_user["role"] != "admin":
        raise HTTPException(status_code=403, detail="需要管理员权限")
    return current_user

# 普通用户端点：需要登录
@app.get("/profile")
async def get_profile(user: Annotated[dict, Depends(get_current_user)]):
    return {"user": user}

# 管理员端点：需要 admin 角色
@app.delete("/admin/users/{user_id}")
async def admin_delete_user(
    user_id: int,
    admin: Annotated[dict, Depends(require_admin)]
):
    return {"deleted": user_id, "by": admin["username"]}

yield 依赖与资源清理

使用 yield 的依赖函数可以在请求处理前后执行代码，用于数据库 Session 管理、资源获取与释放：

from fastapi import FastAPI, Depends
from typing import Generator, AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker

app = FastAPI()

# 创建数据库引擎和会话工厂
engine = create_async_engine("sqlite+aiosqlite:///./app.db")
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

# ── yield 依赖：数据库 Session 管理 ──────────────────────
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """
    yield 前：获取数据库连接
    yield：将 Session 注入到路由函数
    yield 后（finally）：无论是否出错，都关闭连接
    """
    async with AsyncSessionLocal() as session:
        try:
            yield session  # ← 此处暂停，执行路由函数
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        # 离开 with 块时自动调用 session.close()

DBSession = Annotated[AsyncSession, Depends(get_db)]

@app.post("/users/")
async def create_user(username: str, db: DBSession):
    # db 是一个 AsyncSession 对象
    # 路由函数执行完毕后，上面的 finally 块自动关闭连接
    from sqlalchemy import text
    await db.execute(text(f"INSERT INTO users (username) VALUES ('{username}')"))
    return {"created": username}

全局依赖

from fastapi import FastAPI, Depends, Request
import time

# 请求日志依赖（每个请求自动执行）
async def log_request(request: Request):
    print(f"{request.method} {request.url} - start")

# 全局 API Key 验证依赖
async def verify_api_key(x_api_key: Annotated[str, Header()]):
    if x_api_key != "secret-key":
        raise HTTPException(status_code=403, detail="无效 API Key")

# 方式 1：对整个应用添加全局依赖
app = FastAPI(dependencies=[Depends(log_request)])

# 方式 2：对 Router 添加依赖（仅该 Router 下的路由生效）
admin_router = APIRouter(
    prefix="/admin",
    dependencies=[Depends(verify_api_key)]  # 所有 /admin 路由都需要 API Key
)

@admin_router.get("/stats")
async def admin_stats():
    return {"users": 1000, "orders": 5000}