从零开始系统学习 Python 编程

Appearance

03-依赖注入 ​

Python 3.11+

本章讲解 FastAPI 强大的依赖注入系统。

第一部分：依赖注入基础 ​

1.1 实际场景 ​

多个路由需要访问数据库，不想在每个路由中重复创建数据库连接。

问题：如何共享通用逻辑和资源？

1.2 什么是依赖注入？ ​

依赖注入是一种设计模式，允许将依赖项从外部注入到函数或类中。

python
from fastapi import FastAPI, Depends

app: FastAPI = FastAPI()


# 定义依赖
def get_database() -> dict[str, str]:
    return {"connection": "database"}


# 使用依赖
@app.get("/items")
def get_items(db: dict[str, str] = Depends(get_database)) -> dict[str, dict]:
    return {"db": db}

第二部分：函数依赖 ​

2.1 实际场景 ​

分页参数在多个接口中都需要使用，每个接口都需要 skip 和 limit 参数。

问题：如何复用通用参数逻辑？

2.2 基本依赖函数 ​

python
from fastapi import FastAPI, Depends

app: FastAPI = FastAPI()


# 简单依赖
def get_query_param() -> str:
    return "default"


@app.get("/items")
def read_items(q: str = Depends(get_query_param)) -> dict[str, str]:
    return {"query": q}


# 带参数的依赖
def get_pagination(skip: int = 0, limit: int = 10) -> dict[str, int]:
    return {"skip": skip, "limit": limit}


@app.get("/items")
def get_items(pagination: dict[str, int] = Depends(get_pagination)) -> dict[str, int]:
    return pagination

2.3 依赖链 ​

python
def get_database() -> dict[str, str]:
    return {"db": "connected"}


def get_current_user(db: dict[str, str] = Depends(get_database)) -> dict[str, str | int]:
    return {"username": "john", "id": 1}


@app.get("/users/me")
def read_current_user(user: dict[str, str | int] = Depends(get_current_user)) -> dict[str, str | int]:
    return user

第三部分：类依赖 ​

3.1 实际场景 ​

分页逻辑不仅有参数，还有方法如计算偏移量，适合封装成类。

问题：如何将类作为依赖？

3.2 类作为依赖 ​

python
from fastapi import FastAPI, Depends

app: FastAPI = FastAPI()


class Pagination:
    def __init__(self, skip: int = 0, limit: int = 10):
        self.skip: int = skip
        self.limit: int = limit


@app.get("/items")
def get_items(pagination: Pagination = Depends()) -> dict[str, int]:
    return {"skip": pagination.skip, "limit": pagination.limit}

3.3 带方法的类 ​

python
class Database:
    def __init__(self) -> None:
        self.connection: str = "connected"
    
    def query(self, sql: str) -> list[dict[str, str]]:
        return [{"id": 1, "name": "Item 1"}]
    
    def close(self) -> None:
        self.connection = "disconnected"


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


@app.get("/items")
def get_items(db: Database = Depends(get_database)) -> list[dict[str, str]]:
    return db.query("SELECT * FROM items")

第四部分：依赖参数 ​

4.1 实际场景 ​

依赖需要使用路由的路径参数或查询参数。

问题：如何在依赖中访问请求参数？

4.2 路径参数 ​

python
def get_item_id(item_id: int) -> int:
    return item_id


@app.get("/items/{item_id}")
def read_item(item_id: int = Depends(get_item_id)) -> dict[str, int]:
    return {"item_id": item_id}

4.3 查询参数 ​

python
def get_search_query(q: str = "default") -> str:
    return q


@app.get("/search")
def search(q: str = Depends(get_search_query)) -> dict[str, str]:
    return {"query": q}

第五部分：可选依赖 ​

5.1 实际场景 ​

某些接口既支持公开访问，也支持认证访问，认证后返回更多信息。

问题：如何定义可选的依赖？

5.2 可选依赖 ​

python
from typing import Any


def get_optional_db() -> str:
    return "db"


@app.get("/items")
def get_items(db: str | None = Depends(get_optional_db, use_cache=False)) -> dict[str, str | None]:
    return {"db": db}

第六部分：完整示例 ​

python
from fastapi import FastAPI, Depends, Header
from typing import Any

app: FastAPI = FastAPI()


# ==================== 依赖定义 ====================
class Pagination:
    def __init__(self, skip: int = 0, limit: int = 10):
        self.skip: int = skip
        self.limit: int = limit


def get_pagination(skip: int = 0, limit: int = 10) -> Pagination:
    return Pagination(skip=skip, limit=limit)


def verify_token(authorization: str | None = Header(default=None)) -> dict[str, str | int] | None:
    if not authorization:
        return None
    if not authorization.startswith("Bearer "):
        return None
    token: str = authorization.replace("Bearer ", "")
    return {"token": token, "user_id": 1}


def get_current_user(token_data: dict[str, str | int] | None = Depends(verify_token)) -> dict[str, str | int]:
    if not token_data:
        from fastapi import HTTPException
        raise HTTPException(status_code=401, detail="Not authenticated")
    return {"id": token_data["user_id"], "username": "john"}


# ==================== 路由 ====================
@app.get("/public/items")
def get_public_items(pagination: Pagination = Depends(get_pagination)) -> list[dict[str, str]]:
    items: list[dict[str, str]] = [{"id": i, "name": f"Item {i}"} for i in range(1, 101)]
    return items[pagination.skip:pagination.skip + pagination.limit]


@app.get("/private/items")
def get_private_items(
    pagination: Pagination = Depends(get_pagination),
    current_user: dict[str, str | int] = Depends(get_current_user)
) -> dict[str, str | list]:
    items: list[dict[str, str]] = [{"id": i, "name": f"Private Item {i}"} for i in range(1, 51)]
    return {
        "user": current_user["username"],
        "items": items[pagination.skip:pagination.skip + pagination.limit]
    }


@app.get("/profile")
def get_profile(current_user: dict[str, str | int] = Depends(get_current_user)) -> dict[str, str | int]:
    return current_user

第七部分：L3 专家层 ​

7.1 FastAPI 依赖注入的图解析（解决依赖链） ​

FastAPI 的依赖注入系统本质上是一个有向无环图（DAG）解析器。当路由函数声明了依赖，FastAPI 会构建依赖图并按拓扑序执行。

图解析流程：

路由函数: get_user(user=Depends(get_current_user))
                            │
                            ▼
                    ┌───────────────┐
                    │ get_current_  │
                    │     user      │
                    │   Depends:    │
                    │   get_db      │
                    └───────┬───────┘
                            │
                            ▼
                    ┌───────────────┐
                    │    get_db     │
                    │  (叶子节点)    │
                    └───────────────┘

解析后的执行顺序（拓扑排序）：

  Step 1: get_db()         ← 无依赖，最先执行
     │
     ▼
  Step 2: get_current_user(db=result_of_step1)
     │
     ▼
  Step 3: get_user(user=result_of_step2)
     │
     ▼
  返回响应

核心实现（fastapi.dependencies.utils.solve_dependencies）：

python
# 伪代码展示图解析逻辑
async def solve_dependencies(
    *,
    request: Request,
    dependant: Dependant,  # 包含依赖函数的描述
    cache_dependencies: dict[int, Any],  # 依赖缓存
) -> dict[str, Any]:

    values: dict[str, Any] = {}
    sub_dependant: Dependant
    for sub_dependant in dependant.dependencies:
        # 检查缓存
        cache_key = id(sub_dependant.call)
        if cache_key in cache_dependencies:
            values[sub_dependant.name] = cache_dependencies[cache_key]
            continue

        # 递归解析子依赖
        solved = await solve_dependencies(
            request=request,
            dependant=sub_dependant,
            cache_dependencies=cache_dependencies,
        )
        # 调用依赖函数
        result = await run_in_threadpool(sub_dependant.call, **solved)
        values[sub_dependant.name] = result

    return values

依赖图的特点：

  • DAG 结构：不允许循环依赖，FastAPI 在启动时检测并报错
  • 拓扑排序：按依赖关系确定执行顺序，无依赖的先执行
  • 重复依赖合并：同一个依赖函数在图中出现多次时只执行一次（除非 use_cache=False）

7.2 yield 依赖的生命周期管理 ​

使用 yield 的依赖（如数据库会话）由 FastAPI 的生命周期管理机制处理。

生命周期流程：

请求开始
    │
    ▼
┌──────────────────────────────────────┐
│  1. 执行 yield 之前的代码（建立资源）  │
│     db = SessionLocal()              │
└──────────────┬───────────────────────┘
               │
               ▼
┌──────────────────────────────────────┐
│  2. yield 资源给路由函数               │
│     yield db                         │
│     路由函数执行...                    │
└──────────────┬───────────────────────┘
               │
               ▼
┌──────────────────────────────────────┐
│  3. 路由函数完成后                     │
│     a) 正常 → 执行 yield 之后的代码    │
│        try: yield db                 │
│        finally: db.close()  ← 执行   │
│     b) 异常 → 先传播异常再执行清理    │
│        finally 块仍会执行             │
└──────────────────────────────────────┘
               │
               ▼
        响应返回给客户端

底层实现：

  • FastAPI 使用 contextlib.contextmanager 或 contextlib.asynccontextmanager 包装 yield 依赖
  • 依赖被包装为 Dependant 对象，其中 solve 方法处理整个生命周期
  • 异常传播：如果路由抛出异常，FastAPI 会在 finally 块中确保资源被清理，然后让异常继续传播
python
# 内部处理逻辑（简化）
async def solve_yield_dependant(call: Callable) -> AsyncGenerator:
    cm = contextlib.asynccontextmanager(call)()  # 创建异步上下文管理器
    try:
        value = await cm.__aenter__()  # 执行 yield 前代码
        yield value                     # yield 给路由
    except Exception as e:
        await cm.__aexit__(type(e), e, e.__traceback__)  # 传递异常信息
        raise
    else:
        await cm.__aexit__(None, None, None)  # 正常退出

7.3 依赖缓存机制 ​

FastAPI 默认对依赖结果进行缓存，同一请求内的相同依赖只执行一次。

缓存行为：

请求: GET /users/1/items/2

依赖声明:
  /users/{user_id} → Depends(get_user)
  /items/{item_id} → Depends(get_item)
  路由函数         → Depends(get_current_user)
                       └─ Depends(get_user)  ← 相同依赖

执行:
  1. get_user(user_id=1)   ← 执行，结果缓存
  2. get_item(item_id=2)   ← 执行，结果缓存
  3. get_current_user()
     └─ get_user(user_id=1) ← 命中缓存，跳过执行

缓存键: id(dependant.call)  # 依赖函数的内存地址

缓存控制：

  • Depends(func, use_cache=True)：默认值，启用缓存
  • Depends(func, use_cache=False)：每次调用都执行，适用于需要获取最新状态的场景

缓存的作用域： 缓存仅在当前请求内有效，不同请求之间不共享缓存结果。

7.4 性能考量 ​

维度说明建议
图解析启动时构建依赖图，运行时复用依赖深度对性能影响极小
依赖缓存同一请求内默认缓存减少重复的 DB 查询或 API 调用
yield 依赖上下文管理器有微小开销资源管理类依赖应使用 yield
同步依赖放入线程池执行I/O 密集型用 async def 依赖
依赖链深度拓扑排序时间复杂度 O(V+E)实际场景中几乎无影响

7.5 设计动机 ​

设计选择原因
图解析而非简单栈支持复杂的依赖关系，自动解决执行顺序
默认缓存大多数场景下同一依赖在同一请求中只需执行一次
yield 支持资源管理的 Pythonic 方式（RAII 模式）
类型注解驱动利用类型系统进行依赖推断和文档生成

7.6 知识关联 ​

                    依赖注入
                         │
        ┌────────────────┼────────────────┐
        ▼                ▼                ▼
    图解析算法       生命周期管理      缓存机制
    (DAG 拓扑排序)  (yield/finally)  (per-request)
        │                │                │
        ▼                ▼                ▼
    Dependant 对象   上下文管理器     use_cache 参数
    solve_deps()    asynccontextmgr  缓存键 = id(func)
        │                │                │
        └────────────────┼────────────────┘
                         ▼
                  01-FastAPI 入门
                  04-数据库集成 (DB Session)

总结 ​

知识点说明
Depends依赖注入
函数依赖简单依赖函数
类依赖类作为依赖
依赖链依赖嵌套
可选依赖Optional