知识小屋

    依赖注入

    依赖注入是 FastAPI 的核心能力。

    它解决的问题是：多个接口都需要同一段公共逻辑时，不要到处复制。

    常见公共逻辑：

    • 读取分页参数
    • 获取数据库连接
    • 获取当前登录用户
    • 校验 Token
    • 读取配置

    一、最简单的 Depends

    from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


def common_params():
    return {"page": 1, "page_size": 10}


@app.get("/items/")
def list_items(params: Annotated[dict, Depends(common_params)]):
    return params

    访问 /items/ 时，FastAPI 会先执行 common_params()，再把返回值传给 params。

    二、依赖函数也可以接收请求参数

    分页参数经常在多个接口里重复出现。

    from typing import Annotated

from fastapi import Depends, FastAPI, Query

app = FastAPI()


def page_params(
    page: Annotated[int, Query(ge=1)] = 1,
    page_size: Annotated[int, Query(ge=1, le=100)] = 10,
) -> dict[str, int]:
    return {"page": page, "page_size": page_size}


@app.get("/users/")
def list_users(params: Annotated[dict[str, int], Depends(page_params)]):
    return {"module": "users", **params}


@app.get("/items/")
def list_items(params: Annotated[dict[str, int], Depends(page_params)]):
    return {"module": "items", **params}

    访问：

    GET /users/?page=2&page_size=20

    返回：

    {
    "module": "users",
    "page": 2,
    "page_size": 20
}

    三、依赖可以依赖另一个依赖

    比如先读取 Token，再根据 Token 查用户。

    from typing import Annotated

from fastapi import Depends, Header, HTTPException


def get_token(authorization: Annotated[str | None, Header()] = None) -> str:
    if authorization is None:
        raise HTTPException(status_code=401, detail="缺少 Authorization 请求头")
    return authorization


def get_current_user(token: Annotated[str, Depends(get_token)]) -> dict:
    if token != "Bearer demo-token":
        raise HTTPException(status_code=401, detail="Token 无效")
    return {"id": 1, "username": "zhangsan"}

    路由里使用：

    @app.get("/me")
def read_me(user: Annotated[dict, Depends(get_current_user)]):
    return user

    调用时带请求头：

    Authorization: Bearer demo-token

    四、yield 依赖

    有些资源需要“使用前创建，使用后关闭”，例如数据库会话。

    这类依赖用 yield：

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

    执行顺序：

    创建 db
  ↓
进入路由函数使用 db
  ↓
路由结束
  ↓
finally 关闭 db

    路由中使用：

    from typing import Annotated

from fastapi import Depends
from sqlalchemy.orm import Session


@app.get("/users/")
def list_users(db: Annotated[Session, Depends(get_db)]):
    return []

    这里的 SessionLocal 来自 SQLAlchemy 配置。数据库部分可以继续看 SQLAlchemy 教程。

    五、在 APIRouter 上设置依赖

    如果一组接口都需要登录，可以把依赖放到路由上。

    from fastapi import APIRouter, Depends

router = APIRouter(
    prefix="/admin",
    tags=["后台"],
    dependencies=[Depends(get_current_user)],
)


@router.get("/dashboard")
def dashboard():
    return {"message": "后台首页"}

    这样访问 /admin/dashboard 时，也会先执行 get_current_user()。

    如果路由函数里还需要拿到当前用户对象，就把依赖写到参数上：

    @router.get("/profile")
def profile(user: Annotated[dict, Depends(get_current_user)]):
    return user

    六、依赖的缓存

    同一个请求里，同一个依赖默认只会执行一次。

    def get_current_user():
    print("执行一次")
    return {"id": 1}

    如果一个接口里多个地方都依赖它，FastAPI 会复用结果。

    大多数时候这是你想要的行为。只有非常少的场景才需要关闭缓存。

    七、什么时候使用依赖注入

    适合放进依赖的逻辑：

    逻辑是否适合
    获取数据库会话适合
    获取当前登录用户适合
    校验公共请求头适合
    分页参数适合
    具体业务计算不太适合，放到 service 或普通函数更清晰

    依赖注入不是为了把所有代码都变复杂，而是为了让公共逻辑复用得更自然。