4.2 数据库会话与依赖注入

本章目标：写好项目的地基——database.py（连接与会话管理）和 models.py（数据模型），并让 FastAPI 通过依赖注入优雅地使用数据库。

1. app/database.py

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

# 数据库地址。以后换 MySQL 只需要改这一行（并去掉 connect_args）
SQLALCHEMY_DATABASE_URL = "sqlite:///./blog.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},  # 仅 SQLite 需要
    # echo=True,  # 调试时打开可查看 SQL
)

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


class Base(DeclarativeBase):
    """所有模型的基类"""
    pass


def get_db():
    """FastAPI 依赖：每个请求一个独立的数据库会话，请求结束自动关闭"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

get_db 就是第二部分讲的 yield 依赖

回顾执行流程：

请求进来 → SessionLocal() 创建会话
        → yield 把会话交给接口函数使用
        → 接口执行完、响应发出
        → finally 关闭会话（即使接口抛了异常也会关闭）

这实现了黄金原则「一个请求，一个 Session」：不同请求的数据库操作互不干扰，会话绝不泄漏。

定义 SessionDep 类型别名

每个接口都写 db: Annotated[Session, Depends(get_db)] 太啰嗦，在 database.py 末尾加：

from typing import Annotated
from fastapi import Depends

SessionDep = Annotated[Session, Depends(get_db)]

以后接口参数只需写 db: SessionDep。

2. app/models.py

博客的三张表 + 一张中间表（综合运用第三部分的知识）：

from datetime import datetime

from sqlalchemy import Column, ForeignKey, Integer, String, Table, Text, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.database import Base

# 文章-标签 多对多中间表
post_tags = Table(
    "post_tags",
    Base.metadata,
    Column("post_id", Integer, ForeignKey("posts.id"), primary_key=True),
    Column("tag_id", Integer, ForeignKey("tags.id"), primary_key=True),
)


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(100), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    posts: Mapped[list["Post"]] = relationship(
        back_populates="author",
        cascade="all, delete-orphan",   # 删用户时连带删其文章
    )


class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), index=True)
    content: Mapped[str] = mapped_column(Text)
    published: Mapped[bool] = mapped_column(default=False)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), onupdate=func.now()   # 每次更新自动刷新时间
    )

    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    author: Mapped["User"] = relationship(back_populates="posts")

    tags: Mapped[list["Tag"]] = relationship(
        secondary=post_tags, back_populates="posts"
    )


class Tag(Base):
    __tablename__ = "tags"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True, index=True)

    posts: Mapped[list["Post"]] = relationship(
        secondary=post_tags, back_populates="tags"
    )

注意存的是 hashed_password（密码哈希）而不是明文——哈希逻辑第五部分讲，本部分先用一个简化函数占位。

3. app/main.py（初版）

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app import models  # noqa: F401  必须导入，让模型注册到 Base.metadata
from app.database import Base, engine


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时执行：建表（生产环境应改用 Alembic 迁移，见第五部分）
    Base.metadata.create_all(engine)
    yield
    # 关闭时执行：本项目暂时没有需要清理的资源


app = FastAPI(
    title="博客 API",
    description="FastAPI + SQLAlchemy 教程实战项目",
    version="1.0.0",
    lifespan=lifespan,
)


@app.get("/")
def read_root():
    return {"message": "博客 API 正在运行"}

两个要点：

1. lifespan：FastAPI 的生命周期钩子。yield 之前的代码在应用启动时执行一次（我们用来建表），之后的代码在关闭时执行。旧教程里的 @app.on_event("startup") 是过时写法
2. from app import models：看似没用到，实际必不可少——只有导入过 models 模块，那些模型类才会执行、才会登记到 Base.metadata，create_all 才知道要建哪些表。这是新手高频坑：「为什么表没建出来？」多半是忘了导入 models

4. 启动验证

在项目根目录（blog-api/）执行：

fastapi dev app/main.py

• 终端无报错，访问 http://127.0.0.1:8000 看到欢迎信息
• 项目目录下出现 blog.db 文件
• 用 DB Browser for SQLite 打开它，确认 users、posts、tags、post_tags 四张表都在

5. 在接口里使用数据库（预览）

地基打好了，下一章正式写接口。先预览一眼接口层怎么用 SessionDep：

from app.database import SessionDep
from app import models
from sqlalchemy import select


@router.get("/users")
def list_users(db: SessionDep):          # ← 一个参数搞定数据库会话
    return db.scalars(select(models.User)).all()

请求进来时 FastAPI 自动调用 get_db 创建会话注入进来，请求结束自动关闭——你在接口里只管用，完全不用操心开关。

常见问题

Q: 报错 ModuleNotFoundError: No module named 'app'？ 启动命令必须在项目根目录执行（app 文件夹的上一层），且 app/__init__.py 必须存在。

Q: 改了 models.py，数据库没变化？ create_all 不会修改已存在的表。学习阶段删掉 blog.db 重启即可（数据会丢）；正式做法见 Alembic 章节。

Q: 每个请求都新建 Session 会不会慢？ 不会。创建 Session 很轻量，底层连接由 engine 的连接池复用。

本章小结

• database.py：engine 全局一个 + SessionLocal 工厂 + get_db yield 依赖 + SessionDep 别名
• models.py：User/Post/Tag 三模型，含一对多和多对多
• main.py：lifespan 里建表；必须导入 models 否则建不出表
• 「一个请求一个 Session」由依赖注入自动保证

下一章：4.3 用户接口实战 —— schemas、crud、router 三件套。