尧图网站建设 
零基础入门python49FastAPI 数据库初始化与迁移思路一、上一篇课后练习讲解上一篇练习围绕“SQLAlchemy模型与会话”。参考做法是先运行上一篇的测试再用一个成功请求和一个失败请求验证边界本篇在同一项目上增加新能力。上一篇课后练习完整答案上一篇练习已经落实到完整文件运行下面代码可以观察本篇要求的成功和失败状态数据库写入全部放在明确事务边界内。答案要点SQLAlchemy engine、SessionLocal、Base 和 get_db 组成依赖请求结束 finally 关闭 session避免连接泄漏。完整答案文件app/database.py完整参考答案文件完整文件app/database.pyfromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportDeclarativeBase,sessionmaker enginecreate_engine(sqlite:///./blog.db,connect_args{check_same_thread:False})SessionLocalsessionmaker(bindengine,autoflushFalse,expire_on_commitFalse)classBase(DeclarativeBase):passdefget_db():dbSessionLocal()try:yielddbfinally:db.close()验收运行项目测试FastAPI/Flask 使用 python -m pytest -qDjango 使用 python manage.py test再按本文 curl 或 Docker 命令检查预期状态码。完整参考答案文件下面是交付项目中真实存在的完整文件 fastapi-blog/app/database.py。它覆盖本篇新增逻辑以及前文已经完成的依赖代码复制单个函数会丢失上下文因此这里提供整份文件。fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportDeclarativeBase,sessionmaker DATABASE_URLsqlite:///./blog.dbenginecreate_engine(DATABASE_URL,connect_args{check_same_thread:False})SessionLocalsessionmaker(bindengine,autoflushFalse,expire_on_commitFalse)classBase(DeclarativeBase):passdefget_db():dbSessionLocal()try:yielddbfinally:db.close()外键和唯一索引是数据库约束不应只靠 Pydantic。关系属性用于对象导航但列表查询还要考虑 JOIN 和 N1。二、本篇要解决的真实问题create_all 适合教学初始化生产环境应使用 Alembic 记录可回滚迁移本篇演示新增字段时为什么不能直接删表。三、请求是怎样走完整条链路的客户端请求 → 路由匹配 → Pydantic 校验 → Depends 注入用户/Session → SQLAlchemy 查询或业务服务 → 提交事务 → 响应模型序列化。每一步都有明确责任排错时按这个顺序定位。四、先看一个最小代码片段router.get(/api/health)defhealth():return{status:ok}这里的注释不是装饰它说明数据从哪里来、为什么不能相信客户端以及失败时系统应保持什么状态。五、完整文件代码app/database.py当前阶段完整文件fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportDeclarativeBase,sessionmaker DATABASE_URLsqlite:///./blog.dbenginecreate_engine(DATABASE_URL,connect_args{check_same_thread:False})SessionLocalsessionmaker(bindengine,autoflushFalse,expire_on_commitFalse)classBase(DeclarativeBase):passdefget_db():dbSessionLocal()try:yielddbfinally:db.close()阅读方法先找路由函数再找它的 Depends、输入 schema、数据库操作和 response_model这五处合起来才是一个功能。requirements.txt当前阶段完整文件fastapi0.141.1 uvicorn0.52.1 SQLAlchemy2.0.51 PyJWT2.13.0 pwdlib[argon2]0.3.0 python-multipart0.0.22 email-validator2.3.0 httpx0.28.1 pytest9.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-madd article summaryalembic upgrade head alembic current# migrations/versions/20260806_add_summary.pyfromalembicimportopimportsqlalchemyassa revision20260806_add_summarydown_revisionprevious_revisiondefupgrade()-None:# 先允许为空兼容已经存在的旧文章。op.add_column(articles,sa.Column(summary,sa.Text(),nullableTrue))op.create_index(ix_articles_summary,articles,[summary])defdowngrade()-None:op.drop_index(ix_articles_summary,table_namearticles)op.drop_column(articles,summary)自动生成只是候选脚本必须检查是否误删列、旧数据默认值和索引锁表时间。大表变更拆成扩展、回填、收缩三次发布。用空库和含种子数据的临时库各运行 upgrade再执行 downgrade比较表结构和数据是否符合预期。迁移失败时保存完整日志不要手工删除 alembic_version。六、当前项目中的完整文件对照阅读七、一次安全迁移的三阶段以 articles.summary 为例第一版只新增可空列第二版离线回填旧文章第三版再增加非空约束。每一版都能独立回滚部署脚本也能在旧应用和新应用短暂共存时工作。大表回填要分批提交并观察锁等待不能一次 UPDATE 全表。把 revision、执行时间、操作者和数据库环境写入发布记录。开发机上 upgrade 成功不代表生产成功尤其要检查 PostgreSQL 的权限、字符集和索引大小。八、补充代码文件本篇结束完整模块文件本节不是代码片段而是本篇结束时该模块的完整版本。请先备份旧文件再整体替换替换后重新运行本篇命令和测试。阅读时重点看本篇新增的函数、事务边界和错误处理未涉及的代码先不要自行删减。app/database.pyfromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportDeclarativeBase,sessionmaker DATABASE_URLsqlite:///./blog.dbenginecreate_engine(DATABASE_URL,connect_args{check_same_thread:False})SessionLocalsessionmaker(bindengine,autoflushFalse,expire_on_commitFalse)classBase(DeclarativeBase):passdefget_db():dbSessionLocal()try:yielddbfinally:db.close() 零基础入门python49：FastAPI 数据库初始化与迁移思路
零基础入门python49FastAPI 数据库初始化与迁移思路一、上一篇课后练习讲解上一篇练习围绕“SQLAlchemy模型与会话”。参考做法是先运行上一篇的测试再用一个成功请求和一个失败请求验证边界本篇在同一项目上增加新能力。上一篇课后练习完整答案上一篇练习已经落实到完整文件运行下面代码可以观察本篇要求的成功和失败状态数据库写入全部放在明确事务边界内。答案要点SQLAlchemy engine、SessionLocal、Base 和 get_db 组成依赖请求结束 finally 关闭 session避免连接泄漏。完整答案文件app/database.py完整参考答案文件完整文件app/database.pyfromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportDeclarativeBase,sessionmaker enginecreate_engine(sqlite:///./blog.db,connect_args{check_same_thread:False})SessionLocalsessionmaker(bindengine,autoflushFalse,expire_on_commitFalse)classBase(DeclarativeBase):passdefget_db():dbSessionLocal()try:yielddbfinally:db.close()验收运行项目测试FastAPI/Flask 使用 python -m pytest -qDjango 使用 python manage.py test再按本文 curl 或 Docker 命令检查预期状态码。完整参考答案文件下面是交付项目中真实存在的完整文件 fastapi-blog/app/database.py。它覆盖本篇新增逻辑以及前文已经完成的依赖代码复制单个函数会丢失上下文因此这里提供整份文件。fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportDeclarativeBase,sessionmaker DATABASE_URLsqlite:///./blog.dbenginecreate_engine(DATABASE_URL,connect_args{check_same_thread:False})SessionLocalsessionmaker(bindengine,autoflushFalse,expire_on_commitFalse)classBase(DeclarativeBase):passdefget_db():dbSessionLocal()try:yielddbfinally:db.close()外键和唯一索引是数据库约束不应只靠 Pydantic。关系属性用于对象导航但列表查询还要考虑 JOIN 和 N1。二、本篇要解决的真实问题create_all 适合教学初始化生产环境应使用 Alembic 记录可回滚迁移本篇演示新增字段时为什么不能直接删表。三、请求是怎样走完整条链路的客户端请求 → 路由匹配 → Pydantic 校验 → Depends 注入用户/Session → SQLAlchemy 查询或业务服务 → 提交事务 → 响应模型序列化。每一步都有明确责任排错时按这个顺序定位。四、先看一个最小代码片段router.get(/api/health)defhealth():return{status:ok}这里的注释不是装饰它说明数据从哪里来、为什么不能相信客户端以及失败时系统应保持什么状态。五、完整文件代码app/database.py当前阶段完整文件fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportDeclarativeBase,sessionmaker DATABASE_URLsqlite:///./blog.dbenginecreate_engine(DATABASE_URL,connect_args{check_same_thread:False})SessionLocalsessionmaker(bindengine,autoflushFalse,expire_on_commitFalse)classBase(DeclarativeBase):passdefget_db():dbSessionLocal()try:yielddbfinally:db.close()阅读方法先找路由函数再找它的 Depends、输入 schema、数据库操作和 response_model这五处合起来才是一个功能。requirements.txt当前阶段完整文件fastapi0.141.1 uvicorn0.52.1 SQLAlchemy2.0.51 PyJWT2.13.0 pwdlib[argon2]0.3.0 python-multipart0.0.22 email-validator2.3.0 httpx0.28.1 pytest9.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-madd article summaryalembic upgrade head alembic current# migrations/versions/20260806_add_summary.pyfromalembicimportopimportsqlalchemyassa revision20260806_add_summarydown_revisionprevious_revisiondefupgrade()-None:# 先允许为空兼容已经存在的旧文章。op.add_column(articles,sa.Column(summary,sa.Text(),nullableTrue))op.create_index(ix_articles_summary,articles,[summary])defdowngrade()-None:op.drop_index(ix_articles_summary,table_namearticles)op.drop_column(articles,summary)自动生成只是候选脚本必须检查是否误删列、旧数据默认值和索引锁表时间。大表变更拆成扩展、回填、收缩三次发布。用空库和含种子数据的临时库各运行 upgrade再执行 downgrade比较表结构和数据是否符合预期。迁移失败时保存完整日志不要手工删除 alembic_version。六、当前项目中的完整文件对照阅读七、一次安全迁移的三阶段以 articles.summary 为例第一版只新增可空列第二版离线回填旧文章第三版再增加非空约束。每一版都能独立回滚部署脚本也能在旧应用和新应用短暂共存时工作。大表回填要分批提交并观察锁等待不能一次 UPDATE 全表。把 revision、执行时间、操作者和数据库环境写入发布记录。开发机上 upgrade 成功不代表生产成功尤其要检查 PostgreSQL 的权限、字符集和索引大小。八、补充代码文件本篇结束完整模块文件本节不是代码片段而是本篇结束时该模块的完整版本。请先备份旧文件再整体替换替换后重新运行本篇命令和测试。阅读时重点看本篇新增的函数、事务边界和错误处理未涉及的代码先不要自行删减。app/database.pyfromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportDeclarativeBase,sessionmaker DATABASE_URLsqlite:///./blog.dbenginecreate_engine(DATABASE_URL,connect_args{check_same_thread:False})SessionLocalsessionmaker(bindengine,autoflushFalse,expire_on_commitFalse)classBase(DeclarativeBase):passdefget_db():dbSessionLocal()try:yielddbfinally:db.close()