零基础入门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)