零基础入门python47:FastAPI Pydantic请求响应模型
2026/9/3 8:01:06 网站建设 项目流程

零基础入门python47:FastAPI Pydantic请求响应模型

一、上一篇课后练习讲解

上一篇练习围绕“配置、环境变量与依赖注入”。参考做法是先运行上一篇的测试,再用一个成功请求和一个失败请求验证边界;本篇在同一项目上增加新能力。

上一篇课后练习完整答案

上一篇练习已经落实到完整文件,运行下面代码可以观察本篇要求的成功和失败状态;数据库写入全部放在明确事务边界内。
答案要点:Settings 从 .env 读取数据库和 JWT 配置,lru_cache 保证同一进程复用,测试可覆盖配置依赖。

完整答案文件:app/config.py

完整参考答案文件

完整文件:app/config.py

fromfunctoolsimportlru_cachefrompydantic_settingsimportBaseSettingsclassSettings(BaseSettings):database_url:str="sqlite:///./blog.db"jwt_secret:str="dev-change-me"classConfig:env_file=".env"@lru_cachedefget_settings():returnSettings()

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

完整参考答案文件

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

fromfastapiimportFastAPI,Requestfromfastapi.responsesimportJSONResponsefrom.importai,articles,auth,extrasfrom.databaseimportBase,engine Base.metadata.create_all(engine)app=FastAPI(title='Blog Community API',version='2.0')app.include_router(auth.router);app.include_router(articles.router);app.include_router(ai.router);app.include_router(extras.router)@app.get('/api/health')defhealth():return{'status':'ok'}@app.exception_handler(ValueError)defvalue_error_handler(request:Request,exc:ValueError):returnJSONResponse(status_code=400,content={'code':'BAD_REQUEST','message':str(exc)})

jwt_secret没有默认值是故意的:生产缺少密钥时应在启动阶段失败,而不是用公开的开发密钥继续运行。

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

请求模型负责输入校验,响应模型负责稳定输出。422 表示客户端数据没有通过模型约束。

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

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

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

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

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

五、完整文件代码

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;这五处合起来才是一个功能。

app/auth.py(当前阶段完整文件)

fromfastapiimportAPIRouter,Depends,HTTPExceptionfromfastapi.securityimportOAuth2PasswordRequestFormfromsqlalchemyimportselectfromsqlalchemy.ormimportSessionfrom.databaseimportget_dbfrom.modelsimportUserfrom.schemasimportToken,UserCreate,UserReadfrom.securityimportcreate_token,hash_password,verify_password router=APIRouter(prefix='/api/auth',tags=['auth'])@router.post('/register',response_model=UserRead,status_code=201)defregister(body:UserCreate,db:Session=Depends(get_db)):ifdb.scalar(select(User).where(User.email==body.email.lower())):raiseHTTPException(409,'邮箱已注册')user=User(email=body.email.lower(),password_hash=hash_password(body.password))db.add(user);db.commit();db.refresh(user);returnuser@router.post('/login',response_model=Token)deflogin(form:OAuth2PasswordRequestForm=Depends(),db:Session=Depends(get_db)):user=db.scalar(select(User).where(User.email==form.username.lower()))ifnotuserornotverify_password(form.password,user.password_hash):raiseHTTPException(401,'邮箱或密码错误')returnToken(access_token=create_token(user.id))

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

八、课后练习

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

五、Pydantic 不是装饰品:它定义 HTTP 契约

请求模型决定客户端能提交什么,响应模型决定服务端愿意公开什么。ORM 对象中的 password_hash 绝不能出现在用户响应里;密码只在注册请求的内存中出现。

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:strclassArticleCreate(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:str)->str:# 边界清理放在模型,权限和数据库查询仍放在服务层。returnvalue.strip()classArticleRead(ArticleCreate):model_config=ConfigDict(from_attributes=True)id:intauthor_id:intcreated_at:datetime

输入 title=" Python " 应保存为 Python;空标题、过短正文返回 422,业务函数不会执行。from_attributes 只说明如何从 ORM 对象读取属性,不会替你加载关系。需要作者邮箱时,应显式 JOIN 并设计新的响应模型,避免序列化阶段触发 N+1。

curl.exe-i-X POST http://127.0.0.1:8000/api/articles-H"Authorization: Bearer TOKEN"-H"Content-Type: application/json"-d'{"title":"x","content":"short"}'

若返回 500,通常是响应模型字段与 ORM 对象不一致;不要用 Any 掩盖,而应修正契约。把测试用例中的最小长度改坏一次,观察 422 detail 的 loc、msg 和 type 字段。

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

七、不要把校验和业务规则混在一起

“标题不能重复”需要查询数据库,不能写成 Pydantic validator;“只有作者能修改”需要当前用户,也不应放在模型中。模型校验的是单个输入的形状,业务服务校验的是多个对象之间的关系。这个边界能让模型测试保持纯函数,也能让数据库约束成为最终保障。

把错误输入分别记录为字段路径和业务错误码。前端根据 detail 定位表单控件,后端日志则记录 request_id;两者不要共用一段包含数据库细节的字符串。

八、补充代码文件

本篇结束:完整模块文件

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

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)

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询