零基础入门python48FastAPI SQLAlchemy模型与会话一、上一篇课后练习讲解上一篇练习围绕“Pydantic请求响应模型”。参考做法是先运行上一篇的测试再用一个成功请求和一个失败请求验证边界本篇在同一项目上增加新能力。上一篇课后练习完整答案上一篇练习已经落实到完整文件运行下面代码可以观察本篇要求的成功和失败状态数据库写入全部放在明确事务边界内。答案要点Pydantic ArticleCreate 限制标题和正文长度ArticleRead 使用 from_attributes 输出稳定 JSON非法输入自动 422。完整答案文件app/schemas.py完整参考答案文件完整文件app/schemas.pyfrompydanticimportBaseModel,FieldclassArticleCreate(BaseModel):title:strField(min_length1,max_length120)content:strField(min_length1)classArticleRead(ArticleCreate):id:intmodel_config{from_attributes:True}验收运行项目测试FastAPI/Flask 使用 python -m pytest -qDjango 使用 python manage.py test再按本文 curl 或 Docker 命令检查预期状态码。完整参考答案文件下面是交付项目中真实存在的完整文件 fastapi-blog/app/schemas.py。它覆盖本篇新增逻辑以及前文已经完成的依赖代码复制单个函数会丢失上下文因此这里提供整份文件。fromdatetimeimportdatetimefrompydanticimportBaseModel,ConfigDict,EmailStr,Field,field_validatorclassUserCreate(BaseModel):email:EmailStr password:strField(min_length8,max_length128)classUserRead(BaseModel):model_configConfigDict(from_attributesTrue)id:intemail:strclassToken(BaseModel):access_token:strtoken_type:strbearerclassArticleCreate(BaseModel):title:strField(min_length2,max_length120)content:strField(min_length10,max_length20000)field_validator(title)classmethoddefclean_title(cls,value):returnvalue.strip()classArticleRead(ArticleCreate):model_configConfigDict(from_attributesTrue)id:intauthor_id:intcreated_at:datetimeclassCommentCreate(BaseModel):content:strField(min_length1,max_length500)classWritingRequest(BaseModel):content:strField(min_length20,max_length10000)FastAPI 会在进入路由前完成 JSON 解析和字段校验。default_factorylist避免多个请求共享同一个可变默认列表这是 Python 对象生命周期的实际坑。二、本篇要解决的真实问题声明 User、Article 等表Session 负责一次请求的查询和提交请求结束必须关闭 Session。三、请求是怎样走完整条链路的客户端请求 → 路由匹配 → Pydantic 校验 → Depends 注入用户/Session → SQLAlchemy 查询或业务服务 → 提交事务 → 响应模型序列化。每一步都有明确责任排错时按这个顺序定位。四、先看一个最小代码片段router.get(/api/health)defhealth():return{status:ok}这里的注释不是装饰它说明数据从哪里来、为什么不能相信客户端以及失败时系统应保持什么状态。五、完整文件代码app/models.py当前阶段完整文件fromdatetimeimportdatetime,timezonefromsqlalchemyimportForeignKey,String,Text,UniqueConstraintfromsqlalchemy.ormimportMapped,mapped_column,relationshipfrom.databaseimportBaseclassUser(Base):__tablename__usersid:Mapped[int]mapped_column(primary_keyTrue)email:Mapped[str]mapped_column(String(120),uniqueTrue,indexTrue)password_hash:Mapped[str]mapped_column(String(255))created_at:Mapped[datetime]mapped_column(defaultlambda:datetime.now(timezone.utc))classArticle(Base):__tablename__articlesid:Mapped[int]mapped_column(primary_keyTrue)title:Mapped[str]mapped_column(String(120),indexTrue)content:Mapped[str]mapped_column(Text)author_id:Mapped[int]mapped_column(ForeignKey(users.id),indexTrue)created_at:Mapped[datetime]mapped_column(defaultlambda:datetime.now(timezone.utc))comments:Mapped[list[Comment]]relationship(cascadeall, delete-orphan)classComment(Base):__tablename__commentsid:Mapped[int]mapped_column(primary_keyTrue)content:Mapped[str]mapped_column(String(500))article_id:Mapped[int]mapped_column(ForeignKey(articles.id))user_id:Mapped[int]mapped_column(ForeignKey(users.id))classLike(Base):__tablename__likes__table_args__(UniqueConstraint(article_id,user_id),)id:Mapped[int]mapped_column(primary_keyTrue)article_id:Mapped[int]mapped_column(ForeignKey(articles.id))user_id:Mapped[int]mapped_column(ForeignKey(users.id))阅读方法先找路由函数再找它的 Depends、输入 schema、数据库操作和 response_model这五处合起来才是一个功能。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这五处合起来才是一个功能。六、安装、启动与验收..\..\.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 何时提交、何时回滚。能用第二个用户验证资源隔离。能复现一个失败场景并说明原因。八、课后练习围绕“SQLAlchemy模型与会话”新增一个测试用例写出请求、预期响应和断言下一篇开头会给出参考实现并继续使用本项目。五、Session 生命周期和事务边界一个请求拿一个 Session请求结束立即关闭。Session 不是全局连接也不是缓存它跟踪当前事务中的对象。全局 Session 会让并发请求互相看到未提交状态并最终耗尽连接。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():无论路由成功还是异常finally 都关闭当前请求的 Session。dbSessionLocal()try:yielddbfinally:db.close()提交失败必须 rollbacktry:db.add(row)db.commit()exceptIntegrityError:db.rollback()raiseHTTPException(status_code409,detail数据已经存在)在 create_article 中故意在 commit 前抛异常数据库不应出现半条文章在 commit 后抛异常则已经持久化说明函数结束和事务提交不是一回事。SQLite 的 check_same_threadFalse 只是允许测试客户端跨线程访问不代表生产并发配置。六、当前项目中的完整文件对照阅读七、用一个失败事务理解 Session向 likes 表插入重复记录后不要立即再执行查询先 rollback 再查询。可以在 Python shell 中打印 db.in_transaction()观察 commit 前后状态。把异常处理写在最靠近 commit 的位置服务层向上只抛业务可理解的异常。查询列表时使用 db.scalars(stmt).all()单行查询使用 db.scalar 或 db.get不要为了统一而把所有结果都取成 list。返回 ORM 对象前确认 Session 仍然打开否则延迟加载关系会出现 DetachedInstanceError。八、补充代码文件本篇结束完整模块文件本节不是代码片段而是本篇结束时该模块的完整版本。请先备份旧文件再整体替换替换后重新运行本篇命令和测试。阅读时重点看本篇新增的函数、事务边界和错误处理未涉及的代码先不要自行删减。app/models.pyfromdatetimeimportdatetime,timezonefromsqlalchemyimportForeignKey,String,Text,UniqueConstraintfromsqlalchemy.ormimportMapped,mapped_column,relationshipfrom.databaseimportBaseclassUser(Base):__tablename__usersid:Mapped[int]mapped_column(primary_keyTrue)email:Mapped[str]mapped_column(String(120),uniqueTrue,indexTrue)password_hash:Mapped[str]mapped_column(String(255))created_at:Mapped[datetime]mapped_column(defaultlambda:datetime.now(timezone.utc))classArticle(Base):__tablename__articlesid:Mapped[int]mapped_column(primary_keyTrue)title:Mapped[str]mapped_column(String(120),indexTrue)content:Mapped[str]mapped_column(Text)author_id:Mapped[int]mapped_column(ForeignKey(users.id),indexTrue)created_at:Mapped[datetime]mapped_column(defaultlambda:datetime.now(timezone.utc))comments:Mapped[list[Comment]]relationship(cascadeall, delete-orphan)classComment(Base):__tablename__commentsid:Mapped[int]mapped_column(primary_keyTrue)content:Mapped[str]mapped_column(String(500))article_id:Mapped[int]mapped_column(ForeignKey(articles.id))user_id:Mapped[int]mapped_column(ForeignKey(users.id))classLike(Base):__tablename__likes__table_args__(UniqueConstraint(article_id,user_id),)id:Mapped[int]mapped_column(primary_keyTrue)article_id:Mapped[int]mapped_column(ForeignKey(articles.id))user_id:Mapped[int]mapped_column(ForeignKey(users.id)) 零基础入门python48：FastAPI SQLAlchemy模型与会话
零基础入门python48FastAPI SQLAlchemy模型与会话一、上一篇课后练习讲解上一篇练习围绕“Pydantic请求响应模型”。参考做法是先运行上一篇的测试再用一个成功请求和一个失败请求验证边界本篇在同一项目上增加新能力。上一篇课后练习完整答案上一篇练习已经落实到完整文件运行下面代码可以观察本篇要求的成功和失败状态数据库写入全部放在明确事务边界内。答案要点Pydantic ArticleCreate 限制标题和正文长度ArticleRead 使用 from_attributes 输出稳定 JSON非法输入自动 422。完整答案文件app/schemas.py完整参考答案文件完整文件app/schemas.pyfrompydanticimportBaseModel,FieldclassArticleCreate(BaseModel):title:strField(min_length1,max_length120)content:strField(min_length1)classArticleRead(ArticleCreate):id:intmodel_config{from_attributes:True}验收运行项目测试FastAPI/Flask 使用 python -m pytest -qDjango 使用 python manage.py test再按本文 curl 或 Docker 命令检查预期状态码。完整参考答案文件下面是交付项目中真实存在的完整文件 fastapi-blog/app/schemas.py。它覆盖本篇新增逻辑以及前文已经完成的依赖代码复制单个函数会丢失上下文因此这里提供整份文件。fromdatetimeimportdatetimefrompydanticimportBaseModel,ConfigDict,EmailStr,Field,field_validatorclassUserCreate(BaseModel):email:EmailStr password:strField(min_length8,max_length128)classUserRead(BaseModel):model_configConfigDict(from_attributesTrue)id:intemail:strclassToken(BaseModel):access_token:strtoken_type:strbearerclassArticleCreate(BaseModel):title:strField(min_length2,max_length120)content:strField(min_length10,max_length20000)field_validator(title)classmethoddefclean_title(cls,value):returnvalue.strip()classArticleRead(ArticleCreate):model_configConfigDict(from_attributesTrue)id:intauthor_id:intcreated_at:datetimeclassCommentCreate(BaseModel):content:strField(min_length1,max_length500)classWritingRequest(BaseModel):content:strField(min_length20,max_length10000)FastAPI 会在进入路由前完成 JSON 解析和字段校验。default_factorylist避免多个请求共享同一个可变默认列表这是 Python 对象生命周期的实际坑。二、本篇要解决的真实问题声明 User、Article 等表Session 负责一次请求的查询和提交请求结束必须关闭 Session。三、请求是怎样走完整条链路的客户端请求 → 路由匹配 → Pydantic 校验 → Depends 注入用户/Session → SQLAlchemy 查询或业务服务 → 提交事务 → 响应模型序列化。每一步都有明确责任排错时按这个顺序定位。四、先看一个最小代码片段router.get(/api/health)defhealth():return{status:ok}这里的注释不是装饰它说明数据从哪里来、为什么不能相信客户端以及失败时系统应保持什么状态。五、完整文件代码app/models.py当前阶段完整文件fromdatetimeimportdatetime,timezonefromsqlalchemyimportForeignKey,String,Text,UniqueConstraintfromsqlalchemy.ormimportMapped,mapped_column,relationshipfrom.databaseimportBaseclassUser(Base):__tablename__usersid:Mapped[int]mapped_column(primary_keyTrue)email:Mapped[str]mapped_column(String(120),uniqueTrue,indexTrue)password_hash:Mapped[str]mapped_column(String(255))created_at:Mapped[datetime]mapped_column(defaultlambda:datetime.now(timezone.utc))classArticle(Base):__tablename__articlesid:Mapped[int]mapped_column(primary_keyTrue)title:Mapped[str]mapped_column(String(120),indexTrue)content:Mapped[str]mapped_column(Text)author_id:Mapped[int]mapped_column(ForeignKey(users.id),indexTrue)created_at:Mapped[datetime]mapped_column(defaultlambda:datetime.now(timezone.utc))comments:Mapped[list[Comment]]relationship(cascadeall, delete-orphan)classComment(Base):__tablename__commentsid:Mapped[int]mapped_column(primary_keyTrue)content:Mapped[str]mapped_column(String(500))article_id:Mapped[int]mapped_column(ForeignKey(articles.id))user_id:Mapped[int]mapped_column(ForeignKey(users.id))classLike(Base):__tablename__likes__table_args__(UniqueConstraint(article_id,user_id),)id:Mapped[int]mapped_column(primary_keyTrue)article_id:Mapped[int]mapped_column(ForeignKey(articles.id))user_id:Mapped[int]mapped_column(ForeignKey(users.id))阅读方法先找路由函数再找它的 Depends、输入 schema、数据库操作和 response_model这五处合起来才是一个功能。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这五处合起来才是一个功能。六、安装、启动与验收..\..\.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 何时提交、何时回滚。能用第二个用户验证资源隔离。能复现一个失败场景并说明原因。八、课后练习围绕“SQLAlchemy模型与会话”新增一个测试用例写出请求、预期响应和断言下一篇开头会给出参考实现并继续使用本项目。五、Session 生命周期和事务边界一个请求拿一个 Session请求结束立即关闭。Session 不是全局连接也不是缓存它跟踪当前事务中的对象。全局 Session 会让并发请求互相看到未提交状态并最终耗尽连接。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():无论路由成功还是异常finally 都关闭当前请求的 Session。dbSessionLocal()try:yielddbfinally:db.close()提交失败必须 rollbacktry:db.add(row)db.commit()exceptIntegrityError:db.rollback()raiseHTTPException(status_code409,detail数据已经存在)在 create_article 中故意在 commit 前抛异常数据库不应出现半条文章在 commit 后抛异常则已经持久化说明函数结束和事务提交不是一回事。SQLite 的 check_same_threadFalse 只是允许测试客户端跨线程访问不代表生产并发配置。六、当前项目中的完整文件对照阅读七、用一个失败事务理解 Session向 likes 表插入重复记录后不要立即再执行查询先 rollback 再查询。可以在 Python shell 中打印 db.in_transaction()观察 commit 前后状态。把异常处理写在最靠近 commit 的位置服务层向上只抛业务可理解的异常。查询列表时使用 db.scalars(stmt).all()单行查询使用 db.scalar 或 db.get不要为了统一而把所有结果都取成 list。返回 ORM 对象前确认 Session 仍然打开否则延迟加载关系会出现 DetachedInstanceError。八、补充代码文件本篇结束完整模块文件本节不是代码片段而是本篇结束时该模块的完整版本。请先备份旧文件再整体替换替换后重新运行本篇命令和测试。阅读时重点看本篇新增的函数、事务边界和错误处理未涉及的代码先不要自行删减。app/models.pyfromdatetimeimportdatetime,timezonefromsqlalchemyimportForeignKey,String,Text,UniqueConstraintfromsqlalchemy.ormimportMapped,mapped_column,relationshipfrom.databaseimportBaseclassUser(Base):__tablename__usersid:Mapped[int]mapped_column(primary_keyTrue)email:Mapped[str]mapped_column(String(120),uniqueTrue,indexTrue)password_hash:Mapped[str]mapped_column(String(255))created_at:Mapped[datetime]mapped_column(defaultlambda:datetime.now(timezone.utc))classArticle(Base):__tablename__articlesid:Mapped[int]mapped_column(primary_keyTrue)title:Mapped[str]mapped_column(String(120),indexTrue)content:Mapped[str]mapped_column(Text)author_id:Mapped[int]mapped_column(ForeignKey(users.id),indexTrue)created_at:Mapped[datetime]mapped_column(defaultlambda:datetime.now(timezone.utc))comments:Mapped[list[Comment]]relationship(cascadeall, delete-orphan)classComment(Base):__tablename__commentsid:Mapped[int]mapped_column(primary_keyTrue)content:Mapped[str]mapped_column(String(500))article_id:Mapped[int]mapped_column(ForeignKey(articles.id))user_id:Mapped[int]mapped_column(ForeignKey(users.id))classLike(Base):__tablename__likes__table_args__(UniqueConstraint(article_id,user_id),)id:Mapped[int]mapped_column(primary_keyTrue)article_id:Mapped[int]mapped_column(ForeignKey(articles.id))user_id:Mapped[int]mapped_column(ForeignKey(users.id))