零基础入门python53：FastAPI 文章创建与响应模型-平芜编程栈

零基础入门python53：FastAPI 文章创建与响应模型

一、上一篇课后练习讲解

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

上一篇课后练习完整答案

上一篇练习已经落实到完整文件，运行下面代码可以观察本篇要求的成功和失败状态；数据库写入全部放在明确事务边界内。
答案要点：OAuth2PasswordBearer 提供 token，current_user 解码并加载用户；依赖注入让所有受保护路由共享权限边界。

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

完整参考答案文件

完整文件：app/dependencies.py

fromfastapiimportDepends,HTTPExceptionfromfastapi.securityimportOAuth2PasswordBearerfrom.securityimportdecode_token oauth2=OAuth2PasswordBearer(tokenUrl="/auth/login")defcurrent_user(token:str=Depends(oauth2)):user=decode_token(token)ifuserisNone:raiseHTTPException(401,"invalid_token")returnuser

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

完整参考答案文件

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

fromfastapiimportDepends,HTTPExceptionfromfastapi.securityimportOAuth2PasswordBearerfromjwtimportInvalidTokenErrorfromsqlalchemy.ormimportSessionfrom.databaseimportget_dbfrom.modelsimportUserfrom.securityimportdecode_token oauth2_scheme=OAuth2PasswordBearer(tokenUrl='/api/auth/login')defget_current_user(token:str=Depends(oauth2_scheme),db:Session=Depends(get_db)):try:user_id=decode_token(token)except(InvalidTokenError,ValueError):raiseHTTPException(401,'无效或过期令牌')user=db.get(User,user_id)ifnotuser:raiseHTTPException(401,'用户不存在')returnuser

每次请求重新查询用户，是为了让禁用状态立即生效；不要只相信 JWT 中的角色字段。

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

创建文章时作者来自 token 而不是请求体。ArticleRead 用 from_attributes 输出数据库对象，避免手写重复字典。

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

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

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

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

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

五、完整文件代码

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

fromfastapiimportAPIRouter,Depends,HTTPException,Queryfromsqlalchemyimportfunc,or_,selectfromsqlalchemy.excimportIntegrityErrorfromsqlalchemy.ormimportSessionfrom.databaseimportget_dbfrom.dependenciesimportget_current_userfrom.modelsimportArticle,Comment,Like,Userfrom.schemasimportArticleCreate,ArticleRead,CommentCreate router=APIRouter(prefix='/api/articles',tags=['articles'])@router.post('',response_model=ArticleRead,status_code=201)defcreate_article(body:ArticleCreate,user:User=Depends(get_current_user),db:Session=Depends(get_db)):row=Article(**body.model_dump(),author_id=user.id);db.add(row);db.commit();db.refresh(row);returnrow@router.get('')deflist_articles(page:int=Query(1,ge=1),size:int=Query(10,ge=1,le=100),q:str|None=None,db:Session=Depends(get_db)):stmt=select(Article)ifq:stmt=stmt.where(or_(Article.title.contains(q),Article.content.contains(q)))total=db.scalar(select(func.count()).select_from(stmt.subquery()))rows=db.scalars(stmt.order_by(Article.id.desc()).offset((page-1)*size).limit(size)).all()return{'page':page,'size':size,'total':total,'items':[ArticleRead.model_validate(x)forxinrows]}@router.patch('/{article_id}',response_model=ArticleRead)defupdate_article(article_id:int,body:ArticleCreate,user:User=Depends(get_current_user),db:Session=Depends(get_db)):row=db.get(Article,article_id)ifnotrow:raiseHTTPException(404,'文章不存在')ifrow.author_id!=user.id:raiseHTTPException(403,'只能修改自己的文章')row.title,row.content=body.title,body.content;db.commit();db.refresh(row);returnrow@router.delete('/{article_id}',status_code=204)defdelete_article(article_id:int,user:User=Depends(get_current_user),db:Session=Depends(get_db)):row=db.get(Article,article_id)ifnotrow:raiseHTTPException(404,'文章不存在')ifrow.author_id!=user.id:raiseHTTPException(403,'只能删除自己的文章')db.delete(row);db.commit()@router.post('/{article_id}/comments',status_code=201)defcomment(article_id:int,body:CommentCreate,user:User=Depends(get_current_user),db:Session=Depends(get_db)):ifnotdb.get(Article,article_id):raiseHTTPException(404,'文章不存在')row=Comment(content=body.content,article_id=article_id,user_id=user.id)db.add(row);db.commit();db.refresh(row);return{'id':row.id,'content':row.content}@router.post('/{article_id}/likes',status_code=201)deflike(article_id:int,user:User=Depends(get_current_user),db:Session=Depends(get_db)):ifnotdb.get(Article,article_id):raiseHTTPException(404,'文章不存在')db.add(Like(article_id=article_id,user_id=user.id))try:db.commit()exceptIntegrityError:db.rollback();raiseHTTPException(409,'已经点赞')return{'liked':True}

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

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

fromdatetimeimportdatetimefrompydanticimportBaseModel,ConfigDict,EmailStr,Field,field_validatorclassUserCreate(BaseModel):email:EmailStr password:str=Field(min_length=8,max_length=128)classUserRead(BaseModel):model_config=ConfigDict(from_attributes=True)id:intemail:strclassToken(BaseModel):access_token:strtoken_type:str='bearer'classArticleCreate(BaseModel):title:str=Field(min_length=2,max_length=120)content:str=Field(min_length=10,max_length=20000)@field_validator('title')@classmethoddefclean_title(cls,value):returnvalue.strip()classArticleRead(ArticleCreate):model_config=ConfigDict(from_attributes=True)id:intauthor_id:intcreated_at:datetimeclassCommentCreate(BaseModel):content:str=Field(min_length=1,max_length=500)classWritingRequest(BaseModel):content:str=Field(min_length=20,max_length=10000)

阅读方法：先找路由函数，再找它的 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 何时提交、何时回滚。
能用第二个用户验证资源隔离。
能复现一个失败场景并说明原因。

八、课后练习

围绕“文章创建与响应模型”新增一个测试用例，写出请求、预期响应和断言；下一篇开头会给出参考实现，并继续使用本项目。

五、创建文章：作者必须由服务端决定

客户端只能提交标题和正文，不能提交 author_id。作者来自 token 解出的 current_user，这是后端掌握的事实。

@router.post("",response_model=ArticleRead,status_code=201)defcreate_article(body:ArticleCreate,user:User=Depends(get_current_user),db:Session=Depends(get_db)):row=Article(title=body.title,content=body.content,author_id=user.id)db.add(row)db.commit()db.refresh(row)# 读取数据库生成的 id 和 created_atreturnrow

db.refresh 很容易被忽略：自增 id 在提交时才生成。response_model 又会过滤未声明字段。写入失败要回滚并记录上下文，不能捕获异常后返回 200。

发送带 author_id 的 JSON，最终数据库中的作者仍应是 token 对应用户。直接查询 select id,title,author_id from articles order by id desc limit 1，保留查询结果作为本篇证据。

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

七、创建文章的幂等和事务

网络重试可能让客户端重复创建文章。若产品要求幂等，可以让客户端提供 Idempotency-Key，服务端在事务中记录 key 与响应；简单博客也至少应在文档中说明 POST 重试会产生两篇文章。不要用标题去重替代幂等，用户可能合法发布同名文章。

创建文章后再写搜索索引或缓存时，先保证主事务提交成功。附属步骤失败应进入重试队列，不要回滚已经对用户可见的文章。

八、补充代码文件

本篇结束：完整模块文件

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

app/articles.py

fromfastapiimportAPIRouter,Depends,HTTPException,Queryfromsqlalchemyimportfunc,or_,selectfromsqlalchemy.excimportIntegrityErrorfromsqlalchemy.ormimportSessionfrom.databaseimportget_dbfrom.dependenciesimportget_current_userfrom.modelsimportArticle,Comment,Like,Userfrom.schemasimportArticleCreate,ArticleRead,CommentCreate router=APIRouter(prefix='/api/articles',tags=['articles'])@router.post('',response_model=ArticleRead,status_code=201)defcreate_article(body:ArticleCreate,user:User=Depends(get_current_user),db:Session=Depends(get_db)):row=Article(**body.model_dump(),author_id=user.id);db.add(row);db.commit();db.refresh(row);returnrow@router.get('')deflist_articles(page:int=Query(1,ge=1),size:int=Query(10,ge=1,le=100),q:str|None=None,db:Session=Depends(get_db)):stmt=select(Article)ifq:stmt=stmt.where(or_(Article.title.contains(q),Article.content.contains(q)))total=db.scalar(select(func.count()).select_from(stmt.subquery()))rows=db.scalars(stmt.order_by(Article.id.desc()).offset((page-1)*size).limit(size)).all()return{'page':page,'size':size,'total':total,'items':[ArticleRead.model_validate(x)forxinrows]}@router.patch('/{article_id}',response_model=ArticleRead)defupdate_article(article_id:int,body:ArticleCreate,user:User=Depends(get_current_user),db:Session=Depends(get_db)):row=db.get(Article,article_id)ifnotrow:raiseHTTPException(404,'文章不存在')ifrow.author_id!=user.id:raiseHTTPException(403,'只能修改自己的文章')row.title,row.content=body.title,body.content;db.commit();db.refresh(row);returnrow@router.delete('/{article_id}',status_code=204)defdelete_article(article_id:int,user:User=Depends(get_current_user),db:Session=Depends(get_db)):row=db.get(Article,article_id)ifnotrow:raiseHTTPException(404,'文章不存在')ifrow.author_id!=user.id:raiseHTTPException(403,'只能删除自己的文章')db.delete(row);db.commit()@router.post('/{article_id}/comments',status_code=201)defcomment(article_id:int,body:CommentCreate,user:User=Depends(get_current_user),db:Session=Depends(get_db)):ifnotdb.get(Article,article_id):raiseHTTPException(404,'文章不存在')row=Comment(content=body.content,article_id=article_id,user_id=user.id)db.add(row);db.commit();db.refresh(row);return{'id':row.id,'content':row.content}@router.post('/{article_id}/likes',status_code=201)deflike(article_id:int,user:User=Depends(get_current_user),db:Session=Depends(get_db)):ifnotdb.get(Article,article_id):raiseHTTPException(404,'文章不存在')db.add(Like(article_id=article_id,user_id=user.id))try:db.commit()exceptIntegrityError:db.rollback();raiseHTTPException(409,'已经点赞')return{'liked':True}

零基础入门python53：FastAPI 文章创建与响应模型