连接数据库：把课程接口变成真正能保存数据的服务

前面的接口如果把课程放在列表或字典里，运行时看起来没有问题：创建后能查询，修改后也能看到新值。可服务一重启，所有数据就消失了；同时来了两个请求，还可能互相覆盖。数据库接入要解决的，就是把“这次进程记住了什么”变成“系统长期保存了什么”，并且让并发请求在清楚的事务规则下工作。

这一章我们只做一个连续项目：课程管理接口。它从一张 SQLite 表起步，使用 SQLAlchemy 2.x 建模，通过 FastAPI 依赖为每个请求提供独立会话，然后补齐创建、查询、更新、删除、冲突处理、连接池、异步选择、迁移和 PostgreSQL 切换。读完以后，你拿到的不只是几段零散写法，而是一条可以继续扩展的数据库访问链路。

你可以先把图里的几个对象记成一句话：路由处理业务，会话管理一次数据库工作，引擎管理连接，数据库保存最终结果。 后面的代码虽然会逐渐变长，但始终没有离开这条链路。

从 SQLite 的第一张课程表开始

SQLite 很适合这一步，因为它不要求你先安装和管理数据库服务器。一个文件就是一套数据库，删除文件便能重做练习。它不等于生产数据库的缩小版：并发写入能力、类型细节和运维方式都与 PostgreSQL 有差异。不过用它先把模型、会话和事务写对，学习成本最低。

先准备依赖：

python -m pip install fastapi "uvicorn[standard]" sqlalchemy pydantic

我们先把数据库连接和表模型放进 app.py。为了让代码能直接运行，示例暂时使用一个文件；项目增长以后再按 database.py、models.py、schemas.py 和 routers/ 拆分。

from __future__ import annotations
 
import os
from datetime import datetime
 
from fastapi import FastAPI
from sqlalchemy import DateTime, String, create_engine, func
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    mapped_column,
    sessionmaker,
)
 
 
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./courses.db")
 
connect_args = (
    {"check_same_thread": False}
    if DATABASE_URL.startswith("sqlite")
    else {}
)
 
engine = create_engine(
    DATABASE_URL,
    connect_args=connect_args,
    pool_pre_ping=True,
)
 
SessionLocal = sessionmaker(
    bind=engine,
    autoflush=False,
    expire_on_commit=False,
)
 
 
class Base(DeclarativeBase):
    pass
 
 
class Course(Base):
    __tablename__ = "courses"
 
    id: Mapped[int] = mapped_column(primary_key=True)
    code: Mapped[str] = mapped_column(
        String(32), unique=True, index=True
    )
    title: Mapped[str] = mapped_column(String(120))
    level: Mapped[str] = mapped_column(String(20), default="入门")
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.current_timestamp(),
    )
 
 
app = FastAPI(title="课程管理接口")

连接地址里每一段是什么意思

sqlite:///./courses.db 可以拆成三部分看：sqlite 是数据库方言，后面的斜线表达这是文件数据库，./courses.db 是当前目录下的文件。SQLAlchemy 看到这个地址后，会选择 SQLite 对应的驱动，并把 Python 层的操作翻译成 SQLite 能执行的语句。

check_same_thread=False 是 SQLite 驱动相关配置。FastAPI 的同步路由可能在线程池中执行，同一次请求的数据库工作不一定永远停在创建连接的那个线程，因此教程中的文件型 SQLite 通常会关闭这项线程检查。它并不意味着一个 Session 可以被多个请求同时共享；会话仍然必须按请求隔离。

engine 不是一条永久占用的数据库连接。它更像连接与方言的管理入口：需要执行数据库工作时，从池中取得连接；工作结束后，再把连接放回去。应用里通常只创建一个引擎，而不是每个请求都重新创建一个。

SessionLocal 也不是会话本身，它是会话工厂。之后每调用一次 SessionLocal()，才会得到一份新的 Session。

SQLAlchemy 2.x 的类型化模型

现代声明式模型使用 DeclarativeBase、Mapped[...] 和 mapped_column()。Mapped[int] 描述 Python 对象上的属性类型，mapped_column(primary_key=True) 描述数据库列规则。两者放在一起，编辑器能理解属性，SQLAlchemy 也能生成表结构。

这张表有四个值得提前看懂的约束：

• id 是主键。数据库用它唯一定位一行，按主键读取时可以直接调用 db.get(Course, course_id)。
• code 是业务上的课程编码，unique=True 要求全表不能重复。index=True 会建立索引，适合频繁按编码查找。
• title 和 level 限制了字符串长度。长度限制是表结构的一部分，接口层仍应先做更友好的输入校验。
• created_at 使用数据库端默认时间。插入时不必由客户端传值，写入完成后再从数据库取得它。

刚开始练习时，可以运行一次下面的语句建表：

Base.metadata.create_all(bind=engine)

它适合空数据库的第一次启动，却不负责修改已有表。例如你后来给 Course 增加 summary 列，再运行 create_all()，已有的 courses 表不会自动补列。结构开始变化以后，要把这份工作交给迁移脚本，后面会完整处理。

把数据库模型和接口模型分开

第一次写数据库接口时，很容易让一个类同时负责请求校验、响应序列化和数据库映射。单表练习似乎省事，但需求一变就会卡住：创建请求不该带 id，响应里不该泄露内部字段，更新请求又需要所有字段可选。把这些职责拆开，接口才有稳定的边界。

继续在 app.py 中加入三份 Pydantic 模型：

from pydantic import BaseModel, ConfigDict, Field
 
 
class CourseCreate(BaseModel):
    code: str = Field(
        min_length=2,
        max_length=32,
        pattern=r"^[a-z0-9-]+$",
    )
    title: str = Field(min_length=2, max_length=

CourseCreate 面向创建请求。课程编码只允许小写字母、数字和连字符，这样 URL、日志和跨系统同步时更稳定。CourseUpdate 面向局部更新，所以字段可以不传。CourseRead 面向响应，明确告诉调用方会收到哪些字段。

from_attributes=True 很关键。路由返回的是 SQLAlchemy 的 Course 对象，不是普通字典。打开这个配置后，Pydantic 会从对象属性读取 id、code 等字段，再生成响应 JSON。

用 yield 管理每个请求的会话

接下来要把会话交给路由。最危险的写法是先创建一个全局 Session，再让所有请求共用它。Session 会记录已加载对象、待写入改动和当前事务状态，是可变且有状态的对象。两个请求共享它，就等于把两次业务操作塞进同一个账本里。

正确的粒度是：一个同步请求使用一份 Session；一个并发异步任务使用一份 AsyncSession。

from collections.abc import Generator
from typing import Annotated
 
from fastapi import Depends
from sqlalchemy.orm import Session
 
 
def get_db() -> Generator[Session, None, None]:
    with SessionLocal() as session:
        try:
            yield session
        except Exception:
            session.rollback()
            raise
 
 

yield 前面的代码在路由执行前运行，产出的 session 会被注入路由参数；路由结束或抛出异常后，执行权回到依赖中。with SessionLocal() 最终会关闭会话。关闭并不等于删除数据，也不等于关闭整个引擎，它会释放会话持有的资源，并让连接回到池中等待复用。

依赖里的 rollback() 是最后一道清理措施：如果路由里有未处理异常，而当前事务已经开始，就撤销这次尚未提交的工作，然后继续把原异常抛给 FastAPI。对于预期中的数据库冲突，我们仍会在路由里就地捕获、回滚并转换成明确的 HTTP 状态。

这里还要区分“会话作用域”和“事务作用域”。一份会话可以先提交一个事务，然后在再次访问数据库时自动开始下一个事务。Web 接口里最容易理解的约定是：会话覆盖整个请求；一次业务写操作对应一次明确提交或回滚。不要为了少写几行代码，让一次请求里散落着多个没有说明的 commit()。

完成课程的创建、查询、更新和删除

先实现创建接口。它会把通过校验的请求模型转换成 ORM 对象，加入会话，发送待提交改动，提交事务，然后读取数据库最终生成的字段。

from fastapi import HTTPException, status
from sqlalchemy.exc import IntegrityError
 
 
@app.post(
    "/courses",
    response_model=CourseRead,
    status_code=status.HTTP_201_CREATED,
)
def create_course(payload: CourseCreate, db: DbSession) -> Course:
    course = Course(**payload.model_dump())
    db.add(course)
 
    try:
        db.flush()
        db.commit()
    except IntegrityError 

发送下面的请求：

POST /courses
Content-Type: application/json
 
{
  "code": "fastapi-101",
  "title": "FastAPI 入门",
  "level": "入门"
}

接口返回 201 Created：

{
  "id": 1,
  "code": "fastapi-101",
  "title": "FastAPI 入门",
  "level": "入门",
  "created_at": "2026-08-13T03:31:49"
}

再补上列表和单条查询。SQLAlchemy 2.x 推荐使用 select() 构造查询，再由 Session 执行：

from typing import Annotated
 
from fastapi import Query
from sqlalchemy import select
 
 
@app.get("/courses", response_model=list[CourseRead])
def list_courses(
    db: DbSession,
    offset: Annotated[int, Query(ge=0)] = 0,
    limit: Annotated[int, Query(ge=1, 

列表接口限制 limit 最大为 100，避免调用方一次把整张表拖走。稳定分页还需要稳定排序，因此这里明确按 id 排序。db.get() 专门按主键读取，表达比拼装一条过滤语句更直接。

局部更新时，只处理请求里真正出现的字段：

@app.patch("/courses/{course_id}", response_model=CourseRead)
def update_course(
    course_id: int,
    payload: CourseUpdate,
    db: DbSession,
) -> Course:
    course = db.get(Course, course_id)
    if course is None:
        raise HTTPException(status_code=404, detail="课程不存在")
 
    changes = payload.model_dump(

exclude_unset=True 区分了“调用方没传这个字段”和“调用方明确传了一个值”。这正是 PATCH 最常见的语义。如果不用它，模型里的默认 None 也可能被当成改动，意外覆盖原值。

删除接口要同时处理“不存在”和“成功后没有响应体”：

from fastapi import Response
 
 
@app.delete(
    "/courses/{course_id}",
    status_code=status.HTTP_204_NO_CONTENT,
)
def delete_course(course_id: int, db: DbSession) -> Response:
    course = db.get(Course, course_id)
    if course is None:
        raise HTTPException(status_code=404, detail="课程不存在")

删除成功返回 204 No Content，响应体为空。随后再读取同一个 id，会收到：

{
  "detail": "课程不存在"
}

看懂 flush、commit、rollback 和 refresh

这四个动作名称很接近，却处在不同阶段。把它们混在一起，是数据库接口最常见的困惑之一。

db.add(course) 只是把新对象放进会话的待处理集合。此时数据库里未必已经有这一行。

db.flush() 会把待处理改动发送给当前事务中的数据库，但不会提交事务。它很适合提前触发主键生成或约束检查。flush() 成功后，其他事务通常仍看不到未提交的改动；后续若回滚，这些改动会被撤销。

db.commit() 会先处理所有尚未发送的改动，再提交事务。提交成功以后，这次写入才成为数据库里的正式状态，底层连接也可以归还连接池。

db.rollback() 撤销当前事务。更重要的是，一旦数据库语句失败，当前会话通常不能若无其事地继续写；要先回滚，让会话回到可以继续使用的状态。

db.refresh(course) 会针对这个对象重新读取数据库值。它常用于取得数据库生成的字段，例如自增主键、数据库端时间和触发器修改后的内容。它不负责把改动“保存”进去。

如果一次业务操作要创建课程和默认章节，两条写入必须放在同一个事务里。不要先提交课程，再提交章节，否则第二步失败时会留下没有章节的半成品。

def create_course_with_default_lesson(
    payload: CourseCreate,
    db: Session,
) -> Course:
    try:
        course = Course(**payload.model_dump())
        db.add(course)
        db.flush()  # 取得 course.id，但事务尚未提交
 
        lesson = Lesson(
            course_id=course.id,
            title="开始学习",
        )
        db.add(lesson)
 
        db.commit()  # 两条记录一起生效
        db.refresh(course)
        return course
    except

这里的边界非常清楚：课程和默认章节要么同时出现，要么都不出现。事务的价值并不是“所有地方都加一个 commit()”，而是让一组必须共同成功的操作拥有同一个结果。

让数据库守住唯一约束和并发竞争

创建课程前先查一次编码是否存在，能给调用方更早、更友好的提示，但它不能保证正确性。假设请求甲和请求乙几乎同时到达：两者查询时都看不到这个编码，然后都尝试插入。若表上没有唯一约束，两条重复记录就会一起留下。

因此我们把唯一性写在数据库表上：

code: Mapped[str] = mapped_column(
    String(32),
    unique=True,
    index=True,
)

然后在提交处捕获 IntegrityError，回滚事务并返回 409 Conflict。同一个创建请求再次发送时，结果是：

{
  "detail": "课程编码已存在"
}

这套处理分成两层：

• 应用层可以事前查询，用于改善提示或减少明显失败。
• 数据库层必须保留唯一约束，负责在并发竞争中作最终裁决。

不要通过解析整段数据库错误字符串来判断所有冲突。不同数据库、驱动和版本的错误文本可能不同。小项目可以先在确定只有一个唯一约束的接口里统一转换；约束增多后，应根据驱动提供的错误码或约束名，分别映射“课程编码重复”“标题组合冲突”等业务错误。

SQLite 很适合演示约束和事务，但它的并发写入方式与 PostgreSQL 不同。接口准备承受真实并发时，要在目标数据库上重新验证事务隔离、锁等待、超时与重试策略，不能因为 SQLite 测试通过就认定生产行为完全一样。

连接池决定能同时占用多少数据库连接

路由使用的是会话，会话真正执行语句时会向引擎取得连接。连接建立通常比创建 Python 对象昂贵，所以引擎会维护连接池，重复利用已经建立的连接。

切换到 PostgreSQL 后，可以显式配置一组容易观察的参数：

engine = create_engine(
    DATABASE_URL,
    pool_size=5,
    max_overflow=10,
    pool_timeout=30,
    pool_pre_ping=True,
)

这些参数不能孤立地追求“大”：

• pool_size=5 表示池中长期维护的连接规模。
• max_overflow=10 表示高峰期还能临时多开多少连接。
• pool_timeout=30 表示连接都被占用时，等待多久才报超时。
• pool_pre_ping=True 表示借出连接前做存活检查，减少拿到失效连接的概率。

假设应用启动了 4 个工作进程，每个进程都配置 pool_size=5 和 max_overflow=10，理论高峰连接数不是 15，而可能达到 4 × 15 = 60。数据库本身、后台任务和迁移工具也要占连接，所以池大小要从数据库允许的总连接预算倒推。

连接长时间不归还，最常见的原因并不是池太小，而是会话没有关闭、事务跨越了慢网络调用，或者把数据库会话交给脱离请求生命周期的后台任务。一个实用原则是：先完成外部调用，再进入短事务写库；不要拿着数据库连接等待一个不确定何时返回的远程服务。

同步和异步要按整条调用链选择

FastAPI 同时支持普通函数和异步函数，但在路由前加一个 async，不会自动把同步数据库驱动变成异步驱动。真正的异步数据库链路，需要异步引擎、异步驱动和 AsyncSession 配套使用。

同步写法适合这些情况：团队更熟悉同步代码，现有依赖大多是同步库，数据库压力可控，或者服务主要靠多进程与线程池扩展。它更直接，也更容易排查。

异步写法适合调用链本来就是异步的场景，例如一个请求需要等待多个网络资源，且数据库驱动也提供成熟的异步实现。异步的价值是等待 I/O 时让出执行权，不是让一条数据库语句本身“跑得更快”。

PostgreSQL 的异步版本可以这样组织：

from collections.abc import AsyncGenerator
 
from sqlalchemy import select
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
 
 
ASYNC_DATABASE_URL = (
    "postgresql+psycopg://app_user:secret@db:5432/course_db"
)
 
async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    pool_pre_ping=True,
)
 
AsyncSessionLocal = async_sessionmaker(

使用 Psycopg 3 时，同一个 postgresql+psycopg:// 方言可以根据 create_engine() 或 create_async_engine() 选择同步或异步实现。也可以使用 postgresql+asyncpg:// 配合 asyncpg。无论选择哪种驱动，都不要在多个并发任务间共享同一个 AsyncSession；原则仍然是一个任务一份会话。

如果项目已经采用同步 SQLAlchemy，不要仅因为 FastAPI 支持 async def 就仓促改写全部数据层。先测量瓶颈：连接池是否耗尽、慢查询是否缺索引、事务是否过长。很多性能问题应先从 SQL 和事务边界解决，而不是先换编程模型。

用迁移管理结构，再切换到 PostgreSQL

开发初期，Base.metadata.create_all() 能让空数据库快速出现第一张表。进入持续开发后，表结构会变化：增加课程简介、调整字段长度、补索引、拆表。生产数据库里已经有数据，不能删除文件重来，这时需要 Alembic 记录每一次结构变更。

先安装并初始化：

python -m pip install alembic psycopg
alembic init migrations

在 Alembic 的 env.py 中导入模型元数据：

from app import Base
 
target_metadata = Base.metadata

然后生成第一份候选迁移：

alembic revision --autogenerate -m "create courses table"

自动生成不是自动批准。打开新生成的迁移文件，检查 upgrade() 和 downgrade()：表名是否正确、唯一约束是否存在、字段是否会意外丢数据、降级路径是否合理。确认后再执行：

alembic upgrade head

后来给课程增加简介，可以先修改模型：

from sqlalchemy import Text
 
summary: Mapped[str | None] = mapped_column(Text, nullable=True)

再生成并检查下一份迁移：

alembic revision --autogenerate -m "add course summary"
alembic upgrade head

为什么先允许 summary 为空？因为旧记录没有这个值。如果直接添加不可空列，而迁移又没有默认值或回填步骤，已有数据会让升级失败。更稳妥的流程通常是：先加可空列，回填旧数据，最后再改为不可空。

把连接配置换成环境变量

代码已经从 DATABASE_URL 环境变量读取连接地址，因此本地继续使用默认 SQLite，部署时只需提供 PostgreSQL 地址：

export DATABASE_URL='postgresql+psycopg://app_user:secret@db:5432/course_db'
alembic upgrade head
uvicorn app:app --host 0.0.0.0 --port 8000

数据库密码不要写进源码，也不要提交到版本库。生产环境通常由部署平台的密钥配置注入。迁移命令和应用必须读取同一个目标数据库配置，否则很容易出现“迁移了一个库，应用连的是另一个库”。

切换数据库不只是换一行地址，还要逐项确认：

• 安装了连接地址对应的驱动；同步引擎和异步引擎没有混用驱动。
• PostgreSQL 数据库与账号已经创建，账号拥有应用所需的最小权限。
• Alembic 已对目标库执行到最新版本，应用启动时不靠 create_all() 猜结构。
• 连接池总预算考虑了工作进程数量，并给迁移和运维连接留出空间。
• 在 PostgreSQL 上重新运行创建、重复编码、更新、删除、分页和事务回滚测试。
• 针对生产数据量检查查询计划与索引，而不是照搬 SQLite 小数据集下的速度结论。

把整条数据库链路收紧

现在回头看课程创建请求，完整过程是这样的：请求先由 CourseCreate 校验；FastAPI 通过 get_db 创建本次请求的会话；路由构造 Course 映射对象；会话把改动发送给数据库；唯一约束处理并发竞争；提交成功后重新读取数据库生成值；CourseRead 塑造响应；请求结束时会话关闭，连接归还连接池。

这条链路里，每一层都有明确责任：

• Pydantic 模型负责接口输入和输出，不直接决定数据库事务。
• SQLAlchemy 模型负责 Python 对象与表之间的映射，并声明持久化约束。
• Session 负责一次请求中的对象状态和事务，不跨并发请求共享。
• commit()、rollback() 和 refresh() 分别负责生效、撤销与重新读取。
• 数据库唯一约束负责并发情况下的最终正确性。
• 引擎和连接池负责复用连接，但池大小必须服从数据库总预算。
• Alembic 负责让表结构按可审查、可重复的步骤演进。

当你继续增加讲师、章节和选课关系时，不需要推翻这套结构。新增表和关系，补充请求模型，把一次业务操作放进明确事务，再为结构变化写迁移即可。数据库接入真正稳定的标志，不是接口“连上了”，而是每次请求从校验、写入、冲突到资源释放都有可解释的结果。