FastAPI 教程

Chapter 04 · 数据库：数据的持久化家园

📚 学完本章，你将能够：

  • 解释为什么内存列表不能替代真正的数据库
  • 用 SQLAlchemy 2.0 定义数据表模型
  • 执行增删改查（CRUD）操作
  • 理解什么是"数据库会话（Session）"以及如何正确管理它
  • 用 Alembic 管理数据库结构变更（迁移）

💡 场景引入

你的图书 API 目前把所有数据存在 Python 的列表里。服务器重启后，数据全丢了。更可怕的是，两个人同时添加书籍时，列表操作不是线程安全的，可能产生混乱。

数据库是专门用来持久化存储数据的软件。它保证：数据不丢失、多人同时操作不冲突、能高效查询。本章教你用 PostgreSQL + SQLAlchemy 把数据真正存下来。

4.1 数据库 vs 内存列表

特性Python 列表PostgreSQL 数据库
数据持久化❌ 程序结束即消失✅ 写入硬盘，重启仍在
并发安全❌ 多线程可能崩溃✅ 内置锁机制
复杂查询❌ 只能遍历✅ SQL 高效筛选、排序、聚合
事务❌ 没有✅ 要么全成功，要么全回滚
备份恢复❌ 难以备份✅ 成熟工具链

4.2 SQLAlchemy：Python 和数据库的翻译官

你可以直接用 SQL 操作数据库，但那样代码又丑又难维护。ORM（对象关系映射）让你用 Python 类和对象来操作数据库表。

SQLAlchemy 是 Python 最流行的 ORM。本章使用 SQLAlchemy 2.0 语法，它和旧版本有显著不同。

安装依赖
pip install sqlalchemy asyncpg
models.py
from datetime import datetime
from sqlalchemy import String, ForeignKey, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

# 声明式基类，所有模型都继承它
class Base(DeclarativeBase):
    pass

class Book(Base):
    __tablename__ = "books"  # 数据库中的表名

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    author: Mapped[str] = mapped_column(String(100))
    price: Mapped[float]
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

这段代码做了三件事：

  1. DeclarativeBase：声明"我要用类来定义表结构"
  2. Mapped[int]：告诉 Python 类型系统，这个字段是整数
  3. mapped_column(primary_key=True)：告诉 SQLAlchemy，这是主键，数据库会自动自增
Mapped[] 是什么？

它是 SQLAlchemy 2.0 引入的类型标注方式。Mapped[int] 不仅让 IDE 能正确提示，也让 SQLAlchemy 知道这一列在数据库中是什么类型。不写 Mapped 也能工作，但写了更好。

4.3 连接数据库：Engine 和 Session

要操作数据库，你需要两个东西：

database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

# 数据库连接字符串
DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/mydb"

# 创建引擎（连接池）
engine = create_async_engine(DATABASE_URL)

# 创建 Session 工厂
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

# 依赖：每个请求创建一个 Session
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

asyncpg 是 PostgreSQL 的异步驱动。postgresql+asyncpg:// 告诉 SQLAlchemy："用 asyncpg 以异步方式连接 PostgreSQL"。

4.4 CRUD：增删改查

现在把上一章的图书 API 从内存列表迁移到真正的数据库：

创建
@app.post("/books", response_model=BookOut)
async def create_book(data: BookCreate, db: Annotated[AsyncSession, Depends(get_db)]):
    book = Book(title=data.title, author=data.author, price=data.price)
    db.add(book)           # 添加到会话
    await db.commit()      # 提交到数据库
    await db.refresh(book) # 刷新，获取数据库生成的 id
    return book
读取
from sqlalchemy import select

@app.get("/books", response_model=list[BookOut])
async def list_books(db: Annotated[AsyncSession, Depends(get_db)]):
    result = await db.execute(select(Book))
    return result.scalars().all()

@app.get("/books/{book_id}", response_model=BookOut)
async def get_book(book_id: int, db: Annotated[AsyncSession, Depends(get_db)]):
    book = await db.get(Book, book_id)  # 按主键查，简单高效
    if not book:
        raise HTTPException(404, "Book not found")
    return book
更新
@app.put("/books/{book_id}", response_model=BookOut)
async def update_book(book_id: int, data: BookCreate, db: Annotated[AsyncSession, Depends(get_db)]):
    book = await db.get(Book, book_id)
    if not book:
        raise HTTPException(404, "Book not found")
    book.title = data.title
    book.author = data.author
    book.price = data.price
    await db.commit()
    return book
删除
@app.delete("/books/{book_id}")
async def delete_book(book_id: int, db: Annotated[AsyncSession, Depends(get_db)]):
    book = await db.get(Book, book_id)
    if not book:
        raise HTTPException(404, "Book not found")
    await db.delete(book)
    await db.commit()
    return {"message": "Deleted successfully"}

4.5 事务：要么全成功，要么全失败

银行转账是典型的需要事务的场景：A 账户扣款 + B 账户加款，两步必须同时成功。如果第一步成功后服务器崩溃，钱就消失了。

事务示例
@app.post("/transfer")
async def transfer(
    from_id: int, to_id: int, amount: float,
    db: Annotated[AsyncSession, Depends(get_db)]
):
    from_account = await db.get(Account, from_id)
    to_account = await db.get(Account, to_id)

    if from_account.balance < amount:
        raise HTTPException(400, "Insufficient balance")

    from_account.balance -= amount
    to_account.balance += amount

    await db.commit()  # 两步修改一起提交，数据库保证原子性
    return {"message": "Transfer successful"}

如果 commit() 之前的任何一步抛出异常，数据库不会保存任何修改。这就是事务的原子性。

常见陷阱

忘记调用 await db.commit() 是新手最常犯的错误。SQLAlchemy 默认不会自动提交，必须显式调用。只 db.add() 不 commit()，数据不会真正写入数据库。

4.6 Alembic：数据库结构版本控制

开发过程中，你经常需要改表结构：加字段、删字段、建索引。手动改数据库又慢又容易出错，团队协作时更是灾难。Alembic 是 SQLAlchemy 的数据库迁移工具，它把每一次结构变更记录为脚本，可以升级、降级、回滚。

初始化 Alembic
pip install alembic
alembic init alembic

# 修改 alembic.ini 中的 sqlalchemy.url
# 修改 alembic/env.py 导入你的 Base

# 自动生成迁移脚本
alembic revision --autogenerate -m "create books table"

# 执行迁移（升级数据库）
alembic upgrade head

# 查看当前版本
alembic current

# 回退一个版本
alembic downgrade -1

迁移脚本会被保存在 alembic/versions/ 文件夹中，提交到 Git。团队成员拉代码后只需运行 alembic upgrade head 即可同步最新的数据库结构。

🎯 知识检测

问题 1
await db.commit() 的作用是什么？
关闭数据库连接
把当前会话中的修改真正写入数据库
查询数据库最新数据
删除会话中的所有对象
解析：SQLAlchemy 的 Session 是一个"暂存区"。你对对象做的修改（add、delete、改属性）都只存在于内存中，直到调用 commit() 才真正发给数据库。这让你有机会在提交前检查或回滚。
问题 2
为什么用 db.get(Book, book_id) 而不是直接写 SQL？
因为写 SQL 是违法的
ORM 提供类型安全、自动映射和数据库无关性
db.get 运行速度更快
PostgreSQL 不支持手写 SQL
解析：ORM 的最大价值不是性能（有时手写 SQL 更快），而是开发效率和可维护性。它自动把数据库行转成 Python 对象，支持类型检查，且代码不绑定特定数据库（理论上可以从 PostgreSQL 换成 MySQL 而不改业务代码）。
问题 3
事务（Transaction）的核心特性是什么？
让查询结果按字母排序
一组操作要么全部成功，要么全部不生效
自动备份数据库
限制同时连接数据库的用户数量
解析：事务的 ACID 特性中，A（Atomicity，原子性）就是指"全或无"。转账时扣款和加款必须在同一个事务中，否则可能出现钱扣了但对方没收到的情况。

本章小结

✓

数据库保证数据持久和并发安全

内存列表在重启和并发场景下都不堪一击。

✓

SQLAlchemy 是 Python 和 SQL 之间的桥梁

用 Python 类定义表，用 Python 对象操作数据，不用手写 SQL。

✓

Session 是一次数据库对话

add → commit 是两步法。修改只在 commit 后才真正生效。

✓

Alembic 管理数据库结构变更

把改表结构变成可版本控制、可回滚的脚本，团队协作必备。

🛠️ 动手实验

把第 1 章的图书 API 完整迁移到 PostgreSQL：定义 Book 模型、实现 CRUD 五个接口、使用 get_db 依赖管理 Session、配置 Alembic 生成初始迁移并执行。
安装 pgAdmin 或 DBeaver，直观看到数据真的写进了数据库表中。