6.2 依赖注入：管理数据库会话

这是 FastAPI × SQLAlchemy 集成的核心一节。目标：让每个请求自动获得一个 Session、请求结束自动关闭——靠 FastAPI 的杀手锏"依赖注入"实现。

一、问题：每个接口都要 Session，谁来管？

按第 2 章的知识，正确的 Session 用法是"每个工作单元一个，用完就关"。对 Web 应用来说：

一个 HTTP 请求 = 一个工作单元 = 一个 Session。

朴素写法是在每个接口函数里手动开关：

@app.get("/users")
def list_users():
    with SessionLocal() as session:          # 每个接口都写一遍
        return session.scalars(select(User)).all()

能跑，但 50 个接口就要写 50 遍 with，而且事务提交/回滚逻辑散落各处。FastAPI 给了更优雅的方案。

二、依赖注入是什么？

依赖注入（Dependency Injection）：接口函数在参数里声明"我需要一个 XX"，框架负责制造并递给你，用完还负责清理。

def get_db():          # 这是一个"依赖"：负责生产 session
    ...yield session...

@app.get("/users")
def list_users(db: Session = Depends(get_db)):   # 声明：我需要一个 session
    ...直接用 db...                                # FastAPI 造好递进来

好处：资源的"创建-使用-销毁"彻底分离，接口函数只管业务。

三、标准配置：database.py

正式项目把数据库相关的底座集中到一个文件。这是你以后每个 FastAPI 项目都会写的"标准模板"：

# database.py
from typing import Generator

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

# SQLite 专属参数：check_same_thread=False
# （FastAPI 可能在不同线程使用同一连接，SQLite 默认禁止，需关闭该检查）
engine = create_engine(
    "sqlite:///app.db",
    connect_args={"check_same_thread": False},
    echo=True,          # 学习期间开着，上线关掉
)

SessionLocal = sessionmaker(bind=engine, autoflush=False)


class Base(DeclarativeBase):
    pass


def get_db() -> Generator[Session, None, None]:
    """依赖函数：每个请求一个 session，用完自动关闭"""
    with SessionLocal() as session:
        yield session

get_db 的执行时序（理解 yield 依赖）

请求到达
  → FastAPI 调用 get_db()，执行到 yield，把 session 递给接口函数
    → 接口函数用 session 干活、返回结果
  → 响应发出后，回到 yield 之后 → with 退出 → session.close()

yield 前是"准备资源"，yield 后是"清理资源"——像三明治一样把你的接口逻辑夹在中间。即使接口抛异常，清理也一定执行（with 保证）。

想要"自动提交/自动回滚"？升级版：

def get_db() -> Generator[Session, None, None]:
    with SessionLocal() as session:
        try:
            yield session
            session.commit()        # 接口正常结束 → 提交
        except Exception:
            session.rollback()      # 接口抛了异常 → 回滚
            raise

这实现了 5.2 节的原则——"一个请求一个事务，谁代表完整业务谁 commit"。本教程为了教学直观，接口里仍会显式写 commit；两种风格都常见，团队统一即可。

四、在接口中使用

# main.py
from fastapi import Depends, FastAPI
from sqlalchemy import select
from sqlalchemy.orm import Session

from database import Base, engine, get_db
from models import User          # 你的 SQLAlchemy 模型

Base.metadata.create_all(bind=engine)     # 学习期直接建表（正式项目用 Alembic）

app = FastAPI()


@app.get("/users")
def list_users(db: Session = Depends(get_db)):
    users = db.scalars(select(User)).all()
    return [{"id": u.id, "name": u.name} for u in users]


@app.post("/users", status_code=201)
def create_user(name: str, email: str, db: Session = Depends(get_db)):
    user = User(name=name, email=email)
    db.add(user)
    db.commit()
    return {"id": user.id}

db: Session = Depends(get_db) 逐词解读：

• db —— 参数名，随意起
• : Session —— 类型注解，让 IDE 知道它是 SQLAlchemy 会话（自动补全 .scalars 等）
• = Depends(get_db) —— 告诉 FastAPI："这个参数不是客户端传的，调 get_db 生成"

五、更现代的写法：Annotated 类型别名

Depends(get_db) 每个接口都写一遍还是重复。用 Annotated 定义一次别名，处处使用：

# database.py 末尾追加
from typing import Annotated
from fastapi import Depends

DbSession = Annotated[Session, Depends(get_db)]

# 接口签名瞬间清爽：
@app.get("/users")
def list_users(db: DbSession):
    return db.scalars(select(User)).all()


@app.get("/users/{user_id}")
def get_user(user_id: int, db: DbSession):
    ...

这是 FastAPI 官方当前推荐的风格，第 7 章实战全程使用。

六、应用生命周期：lifespan

建表、预热连接等"启动时做一次"的事，规范做法是 lifespan（应用级的 yield 依赖）：

from contextlib import asynccontextmanager
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI):
    Base.metadata.create_all(bind=engine)     # 启动时：建表
    yield                                     # ← 应用运行期间
    engine.dispose()                          # 关闭时：释放连接池


app = FastAPI(lifespan=lifespan)

七、依赖注入的隐藏福利：可测试性

测试时想用内存数据库而不碰真实库？覆盖依赖即可，业务代码零改动：

# test 用的 session 工厂（内存 SQLite）
test_engine = create_engine("sqlite://", connect_args={"check_same_thread": False},
                            poolclass=StaticPool)
TestSession = sessionmaker(bind=test_engine)

def override_get_db():
    with TestSession() as session:
        yield session

app.dependency_overrides[get_db] = override_get_db    # 一行替换！

第 7.3 节写接口测试时会真正用上它。

📝 本节小结

• 原则：一个请求 = 一个 Session，由依赖函数 get_db 统一生产和回收
• get_db 模板：with SessionLocal() as session: yield session——yield 前准备、后清理
• 接口声明 db: Session = Depends(get_db)，或用 Annotated 别名 db: DbSession
• SQLite + FastAPI 记得 connect_args={"check_same_thread": False}
• 启动逻辑放 lifespan；测试用 dependency_overrides 换库

下一节解决"两种模型"的配合问题 → 6.3 Pydantic 与 ORM 模型的配合