企业官网建设流程全解析

零基础入门python49：FastAPI 数据库初始化与迁移思路

一、上一篇课后练习讲解

上一篇练习围绕“SQLAlchemy模型与会话”。参考做法是先运行上一篇的测试，再用一个成功请求和一个失败请求验证边界；本篇在同一项目上增加新能力。

上一篇课后练习完整答案

上一篇练习已经落实到完整文件，运行下面代码可以观察本篇要求的成功和失败状态；数据库写入全部放在明确事务边界内。
答案要点：SQLAlchemy engine、SessionLocal、Base 和 get_db 组成依赖；请求结束 finally 关闭 session，避免连接泄漏。

完整答案文件：app/database.py

完整参考答案文件

完整文件：app/database.py

fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportDeclarativeBase,sessionmaker engine=create_engine("sqlite:///./blog.db",connect_args={"check_same_thread":False})SessionLocal=sessionmaker(bind=engine,autoflush=False,expire_on_commit=False)classBase(DeclarativeBase):passdefget_db():db=SessionLocal()try:yielddbfinally:db.close()

验收：运行项目测试（FastAPI/Flask 使用 python -m pytest -q，Django 使用 python manage.py test），再按本文 curl 或 Docker 命令检查预期状态码。

完整参考答案文件

下面是交付项目中真实存在的完整文件 fastapi-blog/app/database.py。它覆盖本篇新增逻辑以及前文已经完成的依赖代码；复制单个函数会丢失上下文，因此这里提供整份文件。

fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportDeclarativeBase,sessionmaker DATABASE_URL='sqlite:///./blog.db'engine=create_engine(DATABASE_URL,connect_args={'check_same_thread':False})SessionLocal=sessionmaker(bind=engine,autoflush=False,expire_on_commit=False)classBase(DeclarativeBase):passdefget_db():db=SessionLocal()try:yielddbfinally:db.close()

外键和唯一索引是数据库约束，不应只靠 Pydantic。关系属性用于对象导航，但列表查询还要考虑 JOIN 和 N+1。

二、本篇要解决的真实问题

create_all 适合教学初始化，生产环境应使用 Alembic 记录可回滚迁移；本篇演示新增字段时为什么不能直接删表。

三、请求是怎样走完整条链路的

客户端请求 → 路由匹配 → Pydantic 校验 → Depends 注入用户/Session → SQLAlchemy 查询或业务服务 → 提交事务 → 响应模型序列化。每一步都有明确责任，排错时按这个顺序定位。

四、先看一个最小代码片段

@router.get("/api/health")defhealth():return{"status":"ok"}

这里的注释不是装饰：它说明数据从哪里来、为什么不能相信客户端，以及失败时系统应保持什么状态。

五、完整文件代码

app/database.py（当前阶段完整文件）

fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportDeclarativeBase,sessionmaker DATABASE_URL='sqlite:///./blog.db'engine=create_engine(DATABASE_URL,connect_args={'check_same_thread':False})SessionLocal=sessionmaker(bind=engine,autoflush=False,expire_on_commit=False)classBase(DeclarativeBase):passdefget_db():db=SessionLocal()try:yielddbfinally:db.close()

阅读方法：先找路由函数，再找它的 Depends、输入 schema、数据库操作和 response_model；这五处合起来才是一个功能。

requirements.txt（当前阶段完整文件）

fastapi==0.141.1 uvicorn==0.52.1 SQLAlchemy==2.0.51 PyJWT==2.13.0 pwdlib[argon2]==0.3.0 python-multipart==0.0.22 email-validator==2.3.0 httpx==0.28.1 pytest==9.1.1

阅读方法：先找路由函数，再找它的 Depends、输入 schema、数据库操作和 response_model；这五处合起来才是一个功能。

六、安装、启动与验收

..\..\.venv\Scripts\python.exe-m pip install-r requirements.txt..\..\.venv\Scripts\python.exe-m pytest-q..\..\.venv\Scripts\python.exe-m uvicorn app.main:app--reload

打开/docs，按顺序完成注册、登录、创建文章、分页查询、修改删除、评论点赞。错误请求必须看到明确状态码：校验错误 422，未登录 401，无权限 403，重复点赞 409。测试应全部通过。

七、本篇验收清单

• 能指出输入校验发生在哪个 schema。
• 能解释 Session 何时提交、何时回滚。
• 能用第二个用户验证资源隔离。
• 能复现一个失败场景并说明原因。

八、课后练习

围绕“数据库初始化与迁移思路”新增一个测试用例，写出请求、预期响应和断言；下一篇开头会给出参考实现，并继续使用本项目。

五、迁移文件为什么要像源代码一样评审

create_all 适合第一次演示，不适合线上升级。Alembic 用版本表和有序脚本记录结构变化，部署时执行 upgrade head，应用启动不再偷偷改表。

python-m pip install alembic alembic init migrations alembic revision-m"add article summary"alembic upgrade head alembic current

# migrations/versions/20260806_add_summary.pyfromalembicimportopimportsqlalchemyassa revision="20260806_add_summary"down_revision="previous_revision"defupgrade()->None:# 先允许为空，兼容已经存在的旧文章。op.add_column("articles",sa.Column("summary",sa.Text(),nullable=True))op.create_index("ix_articles_summary","articles",["summary"])defdowngrade()->None:op.drop_index("ix_articles_summary",table_name="articles")op.drop_column("articles","summary")

自动生成只是候选脚本，必须检查是否误删列、旧数据默认值和索引锁表时间。大表变更拆成扩展、回填、收缩三次发布。用空库和含种子数据的临时库各运行 upgrade；再执行 downgrade，比较表结构和数据是否符合预期。迁移失败时保存完整日志，不要手工删除 alembic_version。

六、当前项目中的完整文件（对照阅读）

七、一次安全迁移的三阶段

以 articles.summary 为例，第一版只新增可空列，第二版离线回填旧文章，第三版再增加非空约束。每一版都能独立回滚，部署脚本也能在旧应用和新应用短暂共存时工作。大表回填要分批提交并观察锁等待，不能一次 UPDATE 全表。

把 revision、执行时间、操作者和数据库环境写入发布记录。开发机上 upgrade 成功不代表生产成功，尤其要检查 PostgreSQL 的权限、字符集和索引大小。

八、补充代码文件

本篇结束：完整模块文件

本节不是代码片段，而是本篇结束时该模块的完整版本。请先备份旧文件，再整体替换；替换后重新运行本篇命令和测试。阅读时重点看本篇新增的函数、事务边界和错误处理，未涉及的代码先不要自行删减。

app/database.py

fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportDeclarativeBase,sessionmaker DATABASE_URL='sqlite:///./blog.db'engine=create_engine(DATABASE_URL,connect_args={'check_same_thread':False})SessionLocal=sessionmaker(bind=engine,autoflush=False,expire_on_commit=False)classBase(DeclarativeBase):passdefget_db():db=SessionLocal()try:yielddbfinally:db.close()