依赖注入

# ❌ 反面教材：依赖被焊死在函数体里
from storage.cloud import CloudEngine

router = APIRouter(prefix="/articles")

@router.post("")
async def create_article(data: ArticleIn):
    engine = CloudEngine(bucket="inkhub-articles")   # 依赖在这里被写死
    return engine.save(data)

class MemoryEngine: ...      # 内存模拟，测试用
class LocalEngine: ...       # 本地磁盘
class CloudEngine: ...       # 云端对象存储

def publish(data: ArticleIn, engine):     # 依赖从参数递进来
    return engine.save(data)

publish(data, CloudEngine())   # 生产：云端
publish(data, LocalEngine())   # 内网部署：本地
publish(data, MemoryEngine())  # 测试：内存模拟

from fastapi import Depends

# 写法A：默认值挂 Depends（老项目常见）
async def list_articles(store: ArticleStore = Depends(get_engine)):
    ...

# 写法B：Annotated 挂 Depends（官方推荐，FastAPI 0.95+）
from typing import Annotated

async def list_articles(store: Annotated[ArticleStore, Depends(get_engine)]):
    ...

请求进来
   │
   ▼
FastAPI 扫路由签名 → 发现参数挂着 Depends(get_engine)
   │
   ▼
调用 get_engine() ──依赖自己声明的参数（query/path/header），框架照样解析
   │
   ▼
拿到返回值（一个 ArticleStore 实例）
   │
   ▼
塞进 store 参数 → 这才调用你的路由函数

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

app = FastAPI()

def get_engine(
    backend: str = Query(default="memory", pattern="^(memory|local|cloud)$"),
) -> ArticleStore:
    """函数依赖：按 query 参数选引擎"""
    ...

@app.get("/articles")
async def list_articles(store: Annotated[ArticleStore, Depends(get_engine)]) -> dict:
    return {"engine": store.backend, "titles": store.titles()}

# 异步依赖：内部要 await 时用
async def get_engine() -> ArticleStore:
    engine = await load_engine_config()
    return engine

# 同步依赖：纯计算、读配置这类快活儿
def get_page_size(limit: int = Query(default=20, ge=1, le=100)) -> int:
    return limit

@app.get("/articles")
async def list_articles(
    store: Annotated[ArticleStore, Depends(get_engine)],   # 拿到 ArticleStore 实例
    limit: Annotated[int, Depends(get_page_size)],         # 拿到 int
) -> list[str]:
    return store.titles()[:limit]

from fastapi import Header

class PublishGuard:
    """发布权限守卫：类依赖示例"""

    def __init__(self, x_inkhub_editor: str = Header(default="匿名墨友")) -> None:
        self.editor = x_inkhub_editor     # 参数名 → 请求头 X-Inkhub-Editor（下划线转连字符）

    def stamp(self, title: str) -> str:
        return f"《{title}》由 {self.editor} 发布"

@app.post("/articles/publish")
async def publish(
    data: ArticleIn,
    guard: Annotated[PublishGuard, Depends(PublishGuard)],   # 注意：传的是类本身
) -> dict:
    article_id = get_store().write(data.title)
    return {"id": article_id, "stamp": guard.stamp(data.title)}

from typing import Annotated
from fastapi import Depends

# 声明一次：类型 + 依赖打包
StoreDep = Annotated[ArticleStore, Depends(get_engine)]
GuardDep = Annotated[PublishGuard, Depends(PublishGuard)]

# 到处复用，签名干净得像没写过依赖
@app.get("/articles")
async def list_articles(store: StoreDep) -> dict: ...

@app.post("/articles/publish")
async def publish(data: ArticleIn, store: StoreDep, guard: GuardDep) -> dict: ...

@app.delete("/articles/{article_id}")
async def remove(article_id: int, store: StoreDep) -> dict: ...

from collections.abc import AsyncIterator

async def get_db() -> AsyncIterator[AsyncSession]:
    async with session_factory() as session:   # 前置段：开仓借 session
        yield session                            # ⬅ 路由函数在这里跑
    # 后置段：路由跑完，退出 async with，session 自动归还

前置段执行（借）
   │
   ▼
yield 值交给路由 → 路由函数执行 → 响应发回客户端
   │
   ▼
后置段执行（还）

async def lease_store(
    backend: str = Query(default="memory", pattern="^(memory|local|cloud)$"),
) -> AsyncIterator[ArticleStore]:
    store = build_engine(backend)
    print(f"前置段：{store.backend} 就绪")        # 请求进来先执行
    try:
        yield store
    finally:
        print(f"后置段：{store.backend} 收尾归档")  # 响应发完才执行

@app.get("/articles/leased")
async def leased_list(store: Annotated[ArticleStore, Depends(lease_store)]) -> dict:
    return {"engine": store.backend, "titles": store.titles()}

        front_page（父依赖）
        /          \
review_channel   get_engine（子依赖）

def review_channel() -> str:
    return "前沿速递"

def front_page(
    channel: Annotated[str, Depends(review_channel)],   # 依赖套依赖
    store: Annotated[ArticleStore, Depends(get_engine)],
) -> dict[str, str]:
    return {"channel": channel, "headline": store.titles()[0] if store.titles() else "空空如也"}

@app.get("/front-page")
async def read_front_page(page: Annotated[dict[str, str], Depends(front_page)]) -> dict:
    return page

get_current_user            ← 11 章主角
    └── get_token           ← 从请求头解 token
          └── get_settings  ← 读配置

def next_seq() -> int:
    print("next_seq 被调用")     # 观察点
    return random.randint(100, 999)

def dep_left(seq: Annotated[int, Depends(next_seq)]) -> int:
    return seq

def dep_right(seq: Annotated[int, Depends(next_seq)]) -> int:
    return seq

@app.get("/cache/on")
async def cache_on(
    left: Annotated[int, Depends(dep_left)],
    right: Annotated[int, Depends(dep_right)],
) -> dict:
    # left == right：底层 next_seq 只跑了一次，结果被缓存共享
    return {"left": left, "right": right, "same": left == right}

def dep_fresh(seq: Annotated[int, Depends(next_seq, use_cache=False)]) -> int:
    return seq

@app.get("/cache/off")
async def cache_off(
    left: Annotated[int, Depends(dep_left)],
    right: Annotated[int, Depends(dep_fresh)],     # 强制重跑
) -> dict:
    return {"left": left, "right": right, "same": left == right}   # same: False

def editor_gate(
    editor: str = Query(min_length=1, description="作者笔名"),
) -> str:
    return editor

@app.get("/articles/by-me")
async def articles_by_me(
    gate: Annotated[str, Depends(editor_gate)],   # editor 参数从子依赖"透传"进来
) -> dict:
    return {"editor": gate}

def check_channel(channel: str = Query()) -> str: ...
    return channel

@app.get("/articles")
async def list_articles(
    channel: str = Query(description="栏目筛选"),
    ok: Annotated[str, Depends(check_channel)],   # 这里的 channel 与上面的同名同值
) -> dict: ...

class FakeStore(ArticleStore):
    backend = "fake（测试替身）"

    def write(self, title: str) -> int:
        return 42                              # 假数据，随便编

def get_fake_store() -> FakeStore:
    return FakeStore()

# 一行替换，所有用到 Depends(get_engine) 的路由全部生效
app.dependency_overrides[get_engine] = get_fake_store

@app.post("/articles/publish")   # 路由完全不知道自己拿到的是假引擎
async def publish(data: ArticleIn, store: StoreDep) -> dict: ...

from fastapi.testclient import TestClient

def test_publish():
    app.dependency_overrides[get_engine] = get_fake_store     # 换替身
    try:
        with TestClient(app) as client:
            resp = client.post("/articles/publish", json={"title": "测试文"})
            assert resp.status_code == 200
    finally:
        app.dependency_overrides.clear()                      # 必须还回来

# mini_dep.py —— 手搓版 Depends，理解原理用
from collections.abc import Callable
from typing import Any

class MiniDep:
    """迷你注入器：包装一个依赖函数，调用时先查替换表"""

    _overrides: dict[Callable[..., Any], Callable[..., Any]] = {}

    def __init__(self, func: Callable[..., Any]) -> None:
        self.func = func

    def __call__(self, *args: Any, **kwargs: Any) -> Any:
        actual = self._overrides.get(self.func, self.func)   # 有替身用替身
        return actual(*args, **kwargs)

    @classmethod
    def override(cls, original: Callable, replacement: Callable) -> None:
        cls._overrides[original] = replacement               # 登记替身

    @classmethod
    def clear_overrides(cls) -> None:
        cls._overrides.clear()                               # 撤掉全部替身

def get_engine() -> ArticleStore:
    return CloudEngine()

def get_fake_engine() -> ArticleStore:
    return MemoryEngine()

def publish(title: str, engine: ArticleStore = MiniDep(get_engine)) -> int:
    return engine.write(title)        # 不关心引擎是谁，伸手就写

publish("云端首篇")                          # ① 走真云端

MiniDep.override(get_engine, get_fake_engine)
publish("测试第二篇")                        # ② 替身生效，写进内存

MiniDep.clear_overrides()
publish("又回云端")                          # ③ 替身撤干净，恢复原样