第一章：依赖注入核心原理
1.1 依赖树构建机制
PYTHON

from fastapi import Depends


def auth_service():
    return OAuth2Scheme()


def db_conn(auth: dict = Depends(auth_service)):
    return Database(creds=auth)


@app.get("/data")
async def get_data(conn=Depends(db_conn)):
    return conn.query()

依赖树可视化：

graph TD
get_data --> db_conn --> auth_service

1.2 作用域控制
PYTHON

from fastapi import Depends, FastAPI
from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(autocommit=False)


# 请求级作用域
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# 应用级单例
cache = LRUCache(size=100)


def get_cache():
    return cache

第二章：Pydantic深度集成
2.1 动态模型注入
PYTHON

from pydantic import create_model


def dynamic_model(fields: dict):
    return create_model('DynamicModel', **fields)


class FilterFactory:
    @classmethod
    def create(cls, model: BaseModel):
        class QueryParams(model):
            limit: int = 100
            offset: int = 0

        return QueryParams


@app.get("/search")
async def search(params=Depends(FilterFactory.create(User))):
    return params.dict()

2.2 校验逻辑复用
PYTHON

from pydantic import validator, root_validator


class GeoValidator:
    @classmethod
    def lat_validator(cls, v):
        if not -90 <= v <= 90:
            raise ValueError("纬度范围错误")
        return v


class Location(BaseModel):
    lat: float
    lng: float

    _validate_lat = validator('lat', allow_reuse=True)(GeoValidator.lat_validator)

第三章：高级注入模式
3.1 工厂模式注入
PYTHON

class NotificationClient:
    def __init__(self, type: str):
        self.client = self._create_client(type)

    @staticmethod
    def _create_client(type):
        return {
            "sms": SMSClient(),
            "email": EmailClient()
        }[type]


def get_notifier(type: str):
    def _factory():
        return NotificationClient(type)

    return _factory


@app.post("/alert")
async def send_alert(
        notifier: NotificationClient = Depends(get_notifier("sms"))
):
    notifier.client.send()

3.2 条件依赖注入
PYTHON

from fastapi import Header


def feature_flag_dep(feature_name: str):
    class FeatureChecker:
        def __init__(self,
                     enabled: bool = Depends(check_feature_enabled)
                     ):
            if not enabled:
                raise HTTPException(403, "功能未启用")

    return FeatureChecker


def check_feature_enabled(
        feature: str = Header(...),
        config: Config = Depends(get_config)
) -> bool:
    return config.is_enabled(feature)


@app.get("/beta")
async def beta_feature(
        checker=Depends(feature_flag_dep("beta"))
):
    return "功能可用"

第四章：错误处理与调试
4.1 依赖链错误传播
PYTHON

class DatabaseError(Exception):
    pass


def db_dep():
    try:
        yield connection
    except Exception as e:
        raise DatabaseError() from e


@app.exception_handler(DatabaseError)
async def handle_db_error(request, exc):
    return JSONResponse(500, {"detail": "数据库异常"})

4.2 依赖图可视化调试
PYTHON

from fastapi.dependencies.utils import solve_dependencies


def print_dependency_tree():
    routes = app.routes
    for route in routes:
        if isinstance(route, APIRoute):
            solved = solve_dependencies(route.dependant)
            print(f"Route {route.path}:")
            for dep in solved.flat_graph():
                print(f"└─ {dep.call.__name__}")

第五章：测试与维护
5.1 依赖覆盖测试
PYTHON

from fastapi.testclient import TestClient


def override_dep():
    return MockDatabase()


app.dependency_overrides[get_db] = override_dep

client = TestClient(app)
response = client.get("/data")
assert "mock" in response.text

5.2 依赖版本管理
PYTHON

from packaging.version import parse


class VersionedDep:
    def __init__(self, api_version: str = Header(...)):
        self.version = parse(api_version)

    def check_min_version(self, min_version: str):
        if self.version < parse(min_version):
            raise HTTPException(400, "版本过低")


@app.get("/new-feature")
async def new_feature(
        dep: VersionedDep = Depends(),
        checker=Depends(dep.check_min_version("2.3"))
):
    return "功能可用"

课后Quiz
Q1：如何实现跨路由共享查询参数？
A) 在每个路由重复定义参数
B) 使用全局变量存储参数
C) 通过依赖注入共享参数

Q2：依赖注入的yield语句有什么作用？

实现请求后清理逻辑
提高依赖执行速度
支持异步生成器
Q3：如何测试被覆盖的依赖项？

使用dependency_overrides
直接修改源代码
配置环境变量
错误解决方案速查表

架构箴言：优秀的依赖注入设计应遵循SOLID原则，特别是依赖倒置原则（DIP）。建议使用依赖图分析工具保持注入层次不超过5层，对高频依赖实施缓存策略，并定期进行依赖关系审计。

来源：https://blog.cmdragon.cn/posts/3b96477f5460/