一、yield：让函数会"暂停-恢复"

为什么 return 不够用

普通函数用 return 返回结果后，函数就死了——变量释放、执行位置清空，再也没下文。写一个"念诗"函数：

def 念诗():
    return "床前明月光"  # 一句念完，函数结束

想听第二句？对不起，这个函数已经没了。这是 return 的局限：一次性，不可恢复。

数据库 session 的生命周期恰恰不是一次性的。请求进来时创建 session，路由处理完后关闭 session——而且即使中途抛异常也得关。如果用 return，需要在每个路由里手工 try/finally，十个路由就得写十遍。

yield 干了什么

yield 让函数变成生成器：每次遇到 yield，交出值、暂停自己、保留内部状态。下次从暂停点继续跑：

def 念诗():
    yield "床前明月光"   # 交出第一句，暂停
    yield "疑是地上霜"   # 再交第二句，再暂停
    yield "举头望明月"   # 第三句，又暂停
    yield "低头思故乡"   # 最后一句

这不是奇怪的语法糖——这是一个有记忆的函数。暂停时变量还在、执行位置记着，随时可以续命。

底层机制：含 yield 的函数，Python 不直接执行，而是返回一个 generator 对象。外部调用 next() 一次就跑一段，跑到下一个 yield 停住：

诗 = 念诗()           # 不执行，返回 generator
print(next(诗))       # "床前明月光" — 跑到第一个 yield 暂停
print(next(诗))       # "疑是地上霜" — 从暂停处继续
print(next(诗))       # "举头望明月"
print(next(诗))       # "低头思故乡"

FastAPI 利用了这套机制：它检测到依赖函数里有 yield，就自动管理生成器的"推进-暂停"节奏——请求进来跑到 yield 交出 session，路由跑完后回到 yield 之后执行 finally 关闭 session。

return vs yield 一句话：return 交完就死，一次性；yield 交完暂停，随时续命，可多次。

二、get_db 依赖注入：给每个请求递一个 session

为什么不能每个路由自己开 session

三个理由：

• 生命周期难以保证。每个路由自己开 session 就得自己关，异常路径还要包 try/finally，十个路由写十遍，漏一个就是连接泄漏。
• 可替换性为零。路由里硬编码 SessionLocal()，测试时想换成测试数据库？没法换，因为写死了。
• 代码重复。每次 with SessionLocal() as db: 是样板代码，不该出现在业务逻辑里。

get_db 怎么解决

在 dependencies.py 里写一个 yield 生成器，把"创建-交出-关闭"三个动作装在一个函数里。路由只用 Depends(get_db) 声明依赖，FastAPI 自动跑整个生命周期：

# app/api/dependencies.py
from typing import Generator

from sqlalchemy.orm import Session

from app.db.session import SessionLocal


def get_db() -> Generator[Session, None, None]:
    """FastAPI 依赖：每个请求创建会话，结束自动关闭"""
    session = SessionLocal()   # 请求进来 → 创建 session
    try:
        yield session          # 交出 session 给路由，自己暂停等着
    finally:
        session.close()        # 路由结束（正常或异常）→ 必关 session

类型注解 Generator[Session, None, None] 的三个槽位：

• 第一个 Session：yield 交出什么东西——这是唯一需要关心的。
• 第二个 None：外部 send 塞回什么东西——几乎不用，永远 None。
• 第三个 None：函数 return 返回什么——生成器一般不 return。

路由里改造只需一行变化——把 with SessionLocal() as db: 换成函数参数声明：

# app/api/routes/jd.py
from fastapi import Depends
from sqlalchemy.orm import Session

from app.api.dependencies import get_db

@router.get("/records", response_model=PaginatedResponse[JdRecordResponse])
def list_records(
    page: int = Query(1, ge=1),
    size: int = Query(20, ge=1, le=100),
    status: str | None = Query(None),
    db: Session = Depends(get_db),   # ← 就这一行，替代 with SessionLocal()
):
    records = repo.list_with_filter(db, ...)
    ...

Depends(get_db) 在请求进来时触发 get_db()，拿到 yield 交出的 session；请求结束时自动跑到 yield 之后，执行 finally 里的 session.close()。

小结：get_db 一箭三雕——统一生命周期（一处 try/finally 全路由复用）、集中管理、可替换（dependency_overrides 一键切换）。yield 是那个让"请求前后动作"能写在一起的底层机制。

三、测试数据库隔离：别把测试数据写进开发库

为什么测试不能直接连 jd_dev.db

三个致命问题：

• 测试互相污染。test A 插 3 条数据，test B 预期空库查结果为 0——结果被 A 的残留数据干扰，断言失败。测试顺序变了就炸。
• 开发数据可能被测试干掉。哪天你在 jd_dev.db 里存了几条真实分析结果做调试，跑测试时如果有 session.query().delete()，调试数据没了。
• 不可重复。同一个测试跑第二遍，数据库里已经有上次的残留数据，结果不一样。

解决方案：测试用独立的 :memory: 数据库，进程结束自动消失，不碰 jd_dev.db 一根毛。

dependency_overrides：偷梁换柱

FastAPI 的 app.dependency_overrides 是一个字典——key 是原始依赖函数，value 是替代品。测试时把 get_db 替换成返回测试 session 的函数，路由里 Depends(get_db) 拿到的就是测试库的 session：

# tests/conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool

from app.api.dependencies import get_db
from app.db.base import Base
from app.main import app


@pytest.fixture
def db_session() -> Session:
    """每测试一个独立的 :memory: 数据库，跑完自动销毁"""
    engine = create_engine(
        "sqlite:///:memory:",
        echo=False,
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,    # 关键！详见下一节踩坑
    )
    TestingSessionLocal = sessionmaker(bind=engine)
    Base.metadata.create_all(engine)   # 在 :memory: 里建出和 jd_dev.db 一样的表结构

    session = TestingSessionLocal()
    try:
        yield session                  # 交给测试函数
    finally:
        session.close()


@pytest.fixture(autouse=True)
def override_get_db(db_session: Session):
    """每个测试自动把 get_db 换成 db_session"""
    app.dependency_overrides[get_db] = lambda: db_session
    yield
    app.dependency_overrides.clear()    # 测试完清掉，不串下一个测试

autouse=True 意味着每个测试函数自动触发 override_get_db——不用在测试函数参数里声明。yield 前搞替换，yield 后清空 override，干净收尾。

注意 override 的值是 lambda: db_session——一个普通函数，不是生成器。FastAPI 对 override 没有形式要求：原始 get_db 是 yield 生成器，但替代品可以是任何可调用对象，只要返回的东西和路由期望的类型一致（这里是 Session）。这就是"可替换性"的体现——接口契约是 Session，怎么造出来的 FastAPI 不管。

改造后测试函数不再自己开 SessionLocal()，数据改写到 db_session 指向的 :memory: 库：

# tests/test_list_records.py
from sqlalchemy.orm import Session

def _make_record(db: Session, status: str = "success", job_title: str = "测试JD") -> None:
    """插入测试记录 —— 接收 db_session，不自己开 SessionLocal"""
    repo = JdRecordRepository()
    repo.create(db, jd_text="test jd text", job_title=job_title, status=status)
    db.commit()

def test_list_records_with_three_records_returns_all(client: TestClient, db_session: Session):
    _make_record(db_session, status="success")
    _make_record(db_session, status="success", job_title="AI 工程师")
    _make_record(db_session, status="failed")
    response = client.get("/api/v1/jd/records")
    ...

数据流动图：

测试 → _make_record(db_session) ──写──▶ :memory: 数据库
                                             │
client.get("/records") ──Depends(get_db)──▶ override → 同一个 :memory:
                                             
测试结束 → fixture finally 关 session → :memory: 消失，零残留

踩坑：:memory: 是连接级别，不是引擎级别

改造完成后跑 pytest -v，报错：

sqlalchemy.exc.OperationalError: no such table: jd_records

明明 Base.metadata.create_all(engine) 建了表，路由里却报"表不存在"。

现象	真相
db_session fixture 里建了表，但路由查不到	SQLite :memory: 是每个连接独立一个数据库，不是每个引擎
为什么会这样	FastAPI TestClient 把同步路由函数丢进线程池执行。SQLAlchemy 从连接池拿了一个新连接——这个新连接指向的是另一个空的 :memory:
怎么修	poolclass=StaticPool —— 强制所有线程复用同一个连接

StaticPool 让 SQLAlchemy 不建连接池，始终返回同一个连接。所有线程碰的是同一个 :memory:，表当然在。

再加 connect_args={"check_same_thread": False} 允许跨线程访问同一个 SQLite 连接（SQLite 默认禁止，但测试串行执行无并发风险，安全）。

经验：check_same_thread=False 解决"能不能跨线程"的问题，StaticPool 解决"跨线程后能不能看到同一个库"的问题。两个缺一不可。

四、详情接口：列表精简、详情全量

前三节搭好了地基——yield 管生命周期、get_db 管注入、dependency_overrides 管测试隔离。这节把地基用起来：写一个新的详情接口，验证整套机制跑得通。

为什么需要单独的详情接口

列表接口 GET /records 返回的 JdRecordResponse 只含 4 个字段（id / job_title / status / created_at）。但用户点进某一条想看完整内容时，需要 jd_text（JD 原文可能几千字）和 analysis_result（一大坨 JSON 结构化分析结果）。

列表场景用户只需要快速浏览——“哪条成功了、标题是什么”。详情场景才需要全量数据。一个接口服务一个场景，这是 REST API 的职责分离原则。

新建详情 Schema

# app/schemas/common.py
class JdRecordDetailResponse(BaseModel):
    """详情接口返回：比列表多 jd_text、analysis_result、error_message"""
    id: int
    jd_text: str
    analysis_result: dict | None
    job_title: str | None
    status: str
    error_message: str | None
    created_at: datetime

    model_config = {"from_attributes": True}

比列表的 JdRecordResponse 多 3 个字段：jd_text、analysis_result、error_message。from_attributes=True 和列表接口一样，支持 ORM 对象 → Pydantic 的自动转换。

路由实现：路径参数 + 404

# app/api/routes/jd.py
@router.get(
    "/records/{record_id}",
    response_model=JdRecordDetailResponse,
    summary="获取 JD 分析记录详情",
)
def get_record(
    record_id: int,                     # 路径参数，FastAPI 自动提取+类型校验
    db: Session = Depends(get_db),
):
    record = repo.get_by_id(db, record_id)
    if record is None:                  # 查不到 → 404
        raise HTTPException(
            status_code=404,
            detail=f"JD 记录 {record_id} 不存在",
        )
    return JdRecordDetailResponse.model_validate(record)

三点值得注意：

• record_id: int——URL 里的 {record_id} 自动绑定到参数；类型标注 int 后，传非数字 FastAPI 自动返回 422，不用额外写校验。
• 查不到走 404——repo.get_by_id 返回 JdRecord | None，是 None 就 raise HTTPException(404)。这个异常走 3.1 铺好的 http_exception_handler 链路，返回统一 ErrorResponse 格式 {"code": 404, "message": "...", "detail": null}。
• model_validate——ORM 对象转 Pydantic Schema，和列表接口完全一致。

测试

# tests/test_record_detail.py
def test_get_record_returns_full_detail(client: TestClient, db_session: Session):
    """找到记录 → 返回全量字段（含详情特有的 jd_text 等）"""
    record_id = _make_record(db_session, status="success", job_title="Python 工程师")
    response = client.get(f"/api/v1/jd/records/{record_id}")
    assert response.status_code == 200
    data = response.json()
    assert data["job_title"] == "Python 工程师"
    assert "jd_text" in data
    assert "analysis_result" in data

def test_get_record_not_found_returns_404(client: TestClient, db_session: Session):
    """不存在的 ID → 404 + 统一 ErrorResponse 格式"""
    response = client.get("/api/v1/jd/records/99999")
    assert response.status_code == 404
    data = response.json()
    assert data["code"] == 404
    assert "不存在" in data["message"]

def test_get_record_non_numeric_id_returns_422(client: TestClient, db_session: Session):
    """非数字 ID → 422（FastAPI 自动类型校验）"""
    response = client.get("/api/v1/jd/records/abc")
    assert response.status_code == 422

三个场景：找到返回全量、查不到返回 404 + 统一格式、类型不匹配返回 422。所有测试共享 :memory: 数据库，每个测试独立，不污染开发库。

小结：详情接口的本质不是"多写一个路由"，是"列表返回摘要、详情返回正文"的场景分离。路径参数 + 类型标注让 FastAPI 自动处理校验，404 走已有的异常处理链路，零额外配置。

---

全文小结：yield 让函数有了"暂停-恢复"的能力，FastAPI 把这套机制用在依赖注入上，形成了 get_db 的优雅生命周期管理（创建-交出-关闭）。dependency_overrides 是测试数据库隔离的钥匙，StaticPool 是 :memory: 线程安全的关键补丁。列表精简、详情全量——两个接口各自服务不同场景，路径参数 + 类型标注让校验不写自通。pytest 23 passed，全绿。