第五章依赖注入系统

本章目标：理解依赖注入的核心思想，掌握 FastAPI 的 Depends() 机制，能用函数依赖、yield 依赖和类依赖解决实际问题。
预计时长：60 分钟

5.1 什么是依赖注入

核心概念

依赖注入（Dependency Injection）的本质：把「准备工作」从业务逻辑中抽出来，让框架按需自动注入。

一个没有依赖注入的例子：

python

@app.get("/items")
def list_items(skip: int = 0, limit: int = 10, token: str = ""):
    # 每个接口都要重复写认证逻辑
    if token != "secret-token":
        raise HTTPException(status_code=401, detail="未授权")
    return items_db[skip : skip + limit]


@app.get("/users")
def list_users(skip: int = 0, limit: int = 10, token: str = ""):
    # 又写了一遍完全一样的认证逻辑
    if token != "secret-token":
        raise HTTPException(status_code=401, detail="未授权")
    return users_db[skip : skip + limit]

问题很明显：认证逻辑散落在每个接口中，重复且难维护。

用依赖注入重构后：

python

def verify_token(token: str):
    if token != "secret-token":
        raise HTTPException(status_code=401, detail="未授权")
    return token


@app.get("/items")
def list_items(skip: int = 0, limit: int = 10, token: str = Depends(verify_token)):
    return items_db[skip : skip + limit]


@app.get("/users")
def list_users(skip: int = 0, limit: int = 10, token: str = Depends(verify_token)):
    return users_db[skip : skip + limit]

认证逻辑只写一次，所有接口复用。

FastAPI 的 Depends() 机制

请求到达
  │
  ▼
FastAPI 检查路由函数的参数
  │
  ├── 普通参数 → 从路径/查询/请求体提取
  │
  └── Depends(xxx) → 先调用 xxx 函数
        │
        ├── xxx 函数也可以有 Depends → 递归解析
        │
        ├── xxx 执行成功 → 返回值注入到路由函数参数
        │
        └── xxx 抛异常 → 请求直接终止，返回错误响应

关键规则：

规则	说明
依赖是一个可调用对象	函数、类、lambda 都行
依赖的参数和路由参数一样解析	依赖函数也能接收路径参数、查询参数等
依赖可以嵌套	依赖函数内部也能用 `Depends()`
依赖抛异常 = 请求终止	在依赖中 `raise HTTPException` 会直接返回错误

5.2 基础用法

函数作为依赖

最基础的依赖就是一个普通函数：

python

from fastapi import FastAPI, Depends

app = FastAPI(title="依赖注入示例")


def get_db_connection():
    """模拟获取数据库连接"""
    print("获取数据库连接")
    return {"db": "connected"}


@app.get("/items")
def list_items(db=Depends(get_db_connection)):
    return {"message": "获取列表成功", "db_status": db}

每次请求 /items 时，FastAPI 会：

先调用 get_db_connection()
把返回值赋给 db 参数
再执行 list_items() 函数体

公共查询参数提取

多个接口共享相同的分页参数？抽成依赖：

python

from fastapi import FastAPI, Depends, Query

app = FastAPI(title="公共参数示例")

fake_items = [{"name": f"Item {i}"} for i in range(100)]
fake_users = [{"name": f"User {i}"} for i in range(50)]


def common_pagination(
    skip: int = Query(default=0, ge=0, description="跳过条数"),
    limit: int = Query(default=10, ge=1, le=100, description="每页条数"),
):
    return {"skip": skip, "limit": limit}


@app.get("/items")
def list_items(pagination: dict = Depends(common_pagination)):
    skip = pagination["skip"]
    limit = pagination["limit"]
    return fake_items[skip : skip + limit]


@app.get("/users")
def list_users(pagination: dict = Depends(common_pagination)):
    skip = pagination["skip"]
    limit = pagination["limit"]
    return fake_users[skip : skip + limit]

优化前：每个接口各写一套 skip/limit 参数
优化后：分页逻辑集中在 common_pagination，修改一处全局生效

在 /docs 中，skip 和 limit 参数会自动出现在两个接口的查询参数中——对调用者来说完全透明。

多级依赖（依赖的依赖）

依赖可以层层嵌套，FastAPI 自动解析整个依赖链：

python

from fastapi import FastAPI, Depends, HTTPException, Header

app = FastAPI(title="多级依赖示例")


def get_token(authorization: str = Header()):
    """第一层：从请求头提取 Token"""
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="无效的 Authorization 头")
    return authorization.replace("Bearer ", "")


def get_current_user(token: str = Depends(get_token)):
    """第二层：根据 Token 获取用户（依赖第一层）"""
    fake_users = {
        "token-zhangsan": {"id": 1, "name": "张三", "role": "admin"},
        "token-lisi": {"id": 2, "name": "李四", "role": "user"},
    }
    user = fake_users.get(token)
    if not user:
        raise HTTPException(status_code=401, detail="无效的 Token")
    return user


@app.get("/me")
def read_current_user(user: dict = Depends(get_current_user)):
    """第三层：路由函数使用最终结果"""
    return {"message": f"你好, {user['name']}", "user": user}

依赖链的执行过程：

请求: GET /me  Header: Authorization: Bearer token-zhangsan
  │
  ▼
get_token(authorization="Bearer token-zhangsan")
  │ return "token-zhangsan"
  ▼
get_current_user(token="token-zhangsan")
  │ return {"id": 1, "name": "张三", "role": "admin"}
  ▼
read_current_user(user={"id": 1, "name": "张三", "role": "admin"})
  │ return {"message": "你好, 张三", ...}
  ▼
响应: 200 OK

如果任意一层抛异常，后续层不会执行：

请求: GET /me  Header: Authorization: InvalidHeader
  │
  ▼
get_token(authorization="InvalidHeader")
  │ raise HTTPException(401) ← 链在这里断开
  ▼
响应: 401 Unauthorized {"detail": "无效的 Authorization 头"}

5.3 典型应用场景

数据库 Session 管理（yield 依赖）

这是 FastAPI 中最常见的依赖模式。yield 依赖可以在请求处理前后执行代码，非常适合管理需要「打开→使用→关闭」的资源。

python

from fastapi import FastAPI, Depends

app = FastAPI(title="yield 依赖示例")


class FakeDBSession:
    """模拟数据库会话"""

    def __init__(self):
        print("  → 打开数据库连接")

    def query(self, sql: str):
        print(f"  → 执行查询: {sql}")
        return [{"id": 1, "name": "测试数据"}]

    def close(self):
        print("  → 关闭数据库连接")


def get_db():
    """yield 依赖：自动管理数据库会话的生命周期"""
    db = FakeDBSession()
    try:
        yield db
    finally:
        db.close()


@app.get("/users")
def list_users(db: FakeDBSession = Depends(get_db)):
    result = db.query("SELECT * FROM users")
    return result

yield 依赖的执行流程：

请求到达
  │
  ▼
get_db() 执行到 yield ──→ 创建 db，交给路由函数
  │
  ▼
list_users(db) ──→ 使用 db 处理业务
  │
  ▼
请求结束（无论成功还是异常）
  │
  ▼
get_db() 的 finally 块执行 ──→ db.close() 释放资源

对比普通依赖和 yield 依赖：

类型	语法	适合场景
普通依赖	`return value`	计算值、校验、获取配置
yield 依赖	`yield value` ... `finally: cleanup`	数据库连接、文件句柄、锁等需要清理的资源

重要：yield 依赖中 finally 块一定会执行，即使路由函数抛异常也不例外。这保证了资源不会泄漏。

实际项目中的数据库依赖（SQLAlchemy 风格）

python

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session

engine = create_engine("sqlite:///./app.db")
SessionLocal = sessionmaker(bind=engine)


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

这段代码几乎是所有 FastAPI + SQLAlchemy 项目的标配。

获取当前登录用户

结合前面的多级依赖，实现一个完整的用户认证流程：

python

from fastapi import FastAPI, Depends, HTTPException, Header
from pydantic import BaseModel

app = FastAPI(title="用户认证示例")


class User(BaseModel):
    id: int
    username: str
    role: str


FAKE_USERS_DB = {
    "token-admin": User(id=1, username="admin", role="admin"),
    "token-guest": User(id=2, username="guest", role="user"),
}


def get_token(authorization: str = Header()):
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="缺少有效的认证信息")
    return authorization.replace("Bearer ", "")


def get_current_user(token: str = Depends(get_token)) -> User:
    user = FAKE_USERS_DB.get(token)
    if not user:
        raise HTTPException(status_code=401, detail="无效的 Token")
    return user


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


@app.get("/my-items")
def read_my_items(current_user: User = Depends(get_current_user)):
    return {"owner": current_user.username, "items": ["item1", "item2"]}

get_current_user 被多个接口复用，认证逻辑只写一次。

权限校验

在认证的基础上增加权限检查：

python

from fastapi import FastAPI, Depends, HTTPException, Header
from pydantic import BaseModel

app = FastAPI(title="权限校验示例")


class User(BaseModel):
    id: int
    username: str
    role: str


FAKE_USERS_DB = {
    "token-admin": User(id=1, username="admin", role="admin"),
    "token-guest": User(id=2, username="guest", role="user"),
}


def get_token(authorization: str = Header()):
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="缺少有效的认证信息")
    return authorization.replace("Bearer ", "")


def get_current_user(token: str = Depends(get_token)) -> User:
    user = FAKE_USERS_DB.get(token)
    if not user:
        raise HTTPException(status_code=401, detail="无效的 Token")
    return user


def require_admin(current_user: User = Depends(get_current_user)) -> User:
    """在 get_current_user 基础上，额外检查是否为管理员"""
    if current_user.role != "admin":
        raise HTTPException(status_code=403, detail="需要管理员权限")
    return current_user


@app.get("/admin/dashboard")
def admin_dashboard(admin: User = Depends(require_admin)):
    return {"message": f"欢迎管理员 {admin.username}", "secret_data": "仅管理员可见"}


@app.delete("/admin/users/{user_id}")
def delete_user(user_id: int, admin: User = Depends(require_admin)):
    return {"message": f"管理员 {admin.username} 删除了用户 {user_id}"}


@app.get("/profile")
def read_profile(user: User = Depends(get_current_user)):
    return {"message": f"你好 {user.username}", "role": user.role}

依赖链一览：

/profile        → get_current_user → get_token   （普通用户可访问）
/admin/dashboard → require_admin → get_current_user → get_token （仅管理员）
/admin/users/{id} → require_admin → get_current_user → get_token （仅管理员）

测试：

请求	Header	预期
`GET /profile`	`Bearer token-guest`	200 — 普通用户正常访问
`GET /admin/dashboard`	`Bearer token-guest`	403 — 需要管理员权限
`GET /admin/dashboard`	`Bearer token-admin`	200 — 管理员正常访问
`GET /admin/dashboard`	（无 Header）	422 — 缺少 Authorization 头

5.4 类作为依赖

当依赖逻辑比较复杂、需要携带配置参数时，用类比函数更合适。

基本写法

FastAPI 中，类实例化的过程本身就是一次函数调用，因此 类可以直接用作依赖：

python

from fastapi import FastAPI, Depends, Query

app = FastAPI(title="类依赖示例")


class Pagination:
    def __init__(
        self,
        skip: int = Query(default=0, ge=0, description="跳过条数"),
        limit: int = Query(default=10, ge=1, le=100, description="每页条数"),
    ):
        self.skip = skip
        self.limit = limit


fake_items = [{"id": i, "name": f"Item {i}"} for i in range(100)]
fake_users = [{"id": i, "name": f"User {i}"} for i in range(50)]


@app.get("/items")
def list_items(pagination: Pagination = Depends()):
    return fake_items[pagination.skip : pagination.skip + pagination.limit]


@app.get("/users")
def list_users(pagination: Pagination = Depends()):
    return fake_users[pagination.skip : pagination.skip + pagination.limit]

注意：当依赖是类时，可以用 Depends() 不传参数的简写形式，FastAPI 会根据类型注解自动推断。即 pagination: Pagination = Depends() 等价于 pagination: Pagination = Depends(Pagination)。

类依赖 vs 函数依赖

维度	函数依赖	类依赖
语法	`def get_xxx(...)`	`class Xxx: def __init__(self, ...)`
返回值	`return` 一个值	实例本身就是值
属性访问	返回 dict → `result["key"]`	返回实例 → `result.key`（更友好）
适合场景	简单逻辑	复杂逻辑、携带配置

带配置的类依赖

类依赖的真正威力在于可以通过构造参数携带配置：

python

from fastapi import FastAPI, Depends, HTTPException, Header

app = FastAPI(title="带配置的类依赖")


class RoleChecker:
    """可配置的角色检查器"""

    def __init__(self, allowed_roles: list[str]):
        self.allowed_roles = allowed_roles

    def __call__(self, authorization: str = Header()):
        if not authorization.startswith("Bearer "):
            raise HTTPException(status_code=401, detail="缺少认证信息")

        token = authorization.replace("Bearer ", "")

        fake_token_roles = {
            "token-admin": "admin",
            "token-editor": "editor",
            "token-viewer": "viewer",
        }
        role = fake_token_roles.get(token)
        if not role:
            raise HTTPException(status_code=401, detail="无效 Token")
        if role not in self.allowed_roles:
            raise HTTPException(
                status_code=403,
                detail=f"需要以下角色之一: {self.allowed_roles}",
            )
        return {"token": token, "role": role}


allow_admin = RoleChecker(allowed_roles=["admin"])
allow_editor = RoleChecker(allowed_roles=["admin", "editor"])
allow_viewer = RoleChecker(allowed_roles=["admin", "editor", "viewer"])


@app.get("/admin/settings")
def admin_settings(auth: dict = Depends(allow_admin)):
    return {"message": "管理员设置页", "role": auth["role"]}


@app.post("/articles")
def create_article(auth: dict = Depends(allow_editor)):
    return {"message": "文章已创建", "role": auth["role"]}


@app.get("/articles")
def list_articles(auth: dict = Depends(allow_viewer)):
    return {"message": "文章列表", "role": auth["role"]}

核心技巧：实现 __call__ 方法让类实例变成可调用对象。

创建依赖实例（配置阶段）：
  allow_admin = RoleChecker(allowed_roles=["admin"])
  allow_editor = RoleChecker(allowed_roles=["admin", "editor"])
                 ↑ 不同实例，不同配置

请求到达（调用阶段）：
  Depends(allow_admin) → allow_admin(authorization=...) → __call__ 执行

测试：

请求	Header	预期
`GET /articles`	`Bearer token-viewer`	200 — viewer 可查看
`POST /articles`	`Bearer token-viewer`	403 — viewer 不能创建
`POST /articles`	`Bearer token-editor`	200 — editor 可创建
`GET /admin/settings`	`Bearer token-editor`	403 — editor 不能访问管理
`GET /admin/settings`	`Bearer token-admin`	200 — admin 可访问

5.5 动手练习

练习 1：提取公共依赖

下面的代码有重复逻辑，请用依赖注入重构：

python

# 重构前（有重复）
@app.get("/items")
def list_items(skip: int = 0, limit: int = 10, sort_by: str = "id"):
    # 校验 sort_by
    if sort_by not in ["id", "name", "price"]:
        raise HTTPException(status_code=400, detail="无效的排序字段")
    return {"items": "...", "sort_by": sort_by}


@app.get("/orders")
def list_orders(skip: int = 0, limit: int = 10, sort_by: str = "id"):
    # 又校验了一遍
    if sort_by not in ["id", "name", "price"]:
        raise HTTPException(status_code=400, detail="无效的排序字段")
    return {"orders": "...", "sort_by": sort_by}

参考答案：

python

from fastapi import FastAPI, Depends, HTTPException, Query

app = FastAPI()

ALLOWED_SORT_FIELDS = ["id", "name", "price"]


def common_list_params(
    skip: int = Query(default=0, ge=0),
    limit: int = Query(default=10, ge=1, le=100),
    sort_by: str = Query(default="id"),
):
    if sort_by not in ALLOWED_SORT_FIELDS:
        raise HTTPException(status_code=400, detail=f"sort_by 必须是 {ALLOWED_SORT_FIELDS} 之一")
    return {"skip": skip, "limit": limit, "sort_by": sort_by}


@app.get("/items")
def list_items(params: dict = Depends(common_list_params)):
    return {"items": "...", **params}


@app.get("/orders")
def list_orders(params: dict = Depends(common_list_params)):
    return {"orders": "...", **params}

练习 2：实现多级依赖链

实现以下依赖链：get_config → get_db → 路由函数：

python

from fastapi import FastAPI, Depends

app = FastAPI()


def get_config():
    return {"db_url": "sqlite:///./app.db", "debug": True}


def get_db(config: dict = Depends(get_config)):
    print(f"连接数据库: {config['db_url']}")
    db = {"connection": config["db_url"], "status": "connected"}
    return db


@app.get("/status")
def check_status(
    config: dict = Depends(get_config),
    db: dict = Depends(get_db),
):
    return {
        "debug_mode": config["debug"],
        "db_status": db["status"],
    }

启动后访问 /status，观察控制台输出，理解依赖链的执行顺序。

本章小结

概念	要点
依赖注入	把「准备工作」抽成独立函数/类，由框架自动调用并注入结果
`Depends()`	FastAPI 的依赖声明方式，写在路由函数参数中
函数依赖	最基础的形式，`def xxx(...): return value`
多级依赖	依赖函数自身也可以用 `Depends()`，形成链式调用
yield 依赖	`yield value` + `finally: cleanup`，自动管理资源生命周期
类依赖	`__init__` 接收请求参数，`__call__` 实现可配置的依赖逻辑
公共参数提取	分页、排序等重复参数抽成依赖，修改一处全局生效
认证/权限	用依赖链实现 Token 提取 → 用户获取 → 角色校验

下一章预告：我们将学习中间件与异常处理——如何统一捕获和处理错误，以及如何用中间件拦截所有请求实现日志记录、CORS 跨域等功能。

第五章 依赖注入系统 ​

5.1 什么是依赖注入 ​

核心概念 ​

FastAPI 的 Depends() 机制 ​

5.2 基础用法 ​

函数作为依赖 ​

公共查询参数提取 ​

多级依赖（依赖的依赖） ​

5.3 典型应用场景 ​

数据库 Session 管理（yield 依赖） ​

实际项目中的数据库依赖（SQLAlchemy 风格） ​

获取当前登录用户 ​

权限校验 ​

5.4 类作为依赖 ​

基本写法 ​

类依赖 vs 函数依赖 ​

带配置的类依赖 ​

5.5 动手练习 ​

练习 1：提取公共依赖 ​

练习 2：实现多级依赖链 ​

本章小结 ​

第五章依赖注入系统

5.1 什么是依赖注入

核心概念

FastAPI 的 Depends() 机制

5.2 基础用法

函数作为依赖

公共查询参数提取

多级依赖（依赖的依赖）

5.3 典型应用场景

数据库 Session 管理（yield 依赖）

实际项目中的数据库依赖（SQLAlchemy 风格）

获取当前登录用户

权限校验

5.4 类作为依赖

基本写法

类依赖 vs 函数依赖

带配置的类依赖

5.5 动手练习

练习 1：提取公共依赖

练习 2：实现多级依赖链

本章小结