FastAPI 知识点总结 - 第二部分：进阶特性

FastAPI 学习总结 - 第二部分进阶特性六、数据验证Pydantic 模型Pydantic 是 FastAPI 的核心数据验证库使用 Python 类型提示进行数据验证和序列化。基本用法frompydanticimportBaseModel,FieldfromtypingimportOptional,ListclassUser(BaseModel):用户模型username:strField(...,description用户名,min_length3,max_length50)email:strField(...,description邮箱)age:intField(18,description年龄,ge0,le120)is_active:boolField(True,description是否活跃)tags:List[str]Field([],description标签列表)bio:Optional[str]Field(None,description个人简介,max_length500)Field 参数说明参数说明示例default默认值Field(18)...必填字段Field(...)ge大于等于Field(..., ge0)gt大于Field(..., gt0)le小于等于Field(..., le100)lt小于Field(..., lt100)min_length最小长度Field(..., min_length3)max_length最大长度Field(..., max_length50)pattern正则匹配Field(..., pattern^[a-z]$)description字段描述Field(..., description用户名)自定义验证器使用validator装饰器添加自定义验证逻辑frompydanticimportBaseModel,validatorclassUser(BaseModel):username:stremail:strvalidator(username)defusername_must_be_alphanumeric(cls,v):验证用户名只能包含字母和数字ifnotv.isalnum():raiseValueError(用户名只能包含字母和数字)returnvvalidator(email)defemail_must_contain_at(cls,v):验证邮箱格式ifnotinv:raiseValueError(邮箱格式不正确)returnv查询参数校验使用Query校验查询参数fromfastapiimportQueryapp.get(/search)defsearch(q:strQuery(None,description搜索关键词,min_length3,max_length50,regex^[a-zA-Z0-9]$),page:intQuery(1,description页码,ge1),page_size:intQuery(10,description每页数量,ge1,le100)): 搜索接口 :param q: 搜索关键词 :param page: 页码默认1 :param page_size: 每页数量默认10 :return: 搜索结果 return{q:q,page:page,page_size:page_size,results:[]}路径参数校验使用Path校验路径参数fromfastapiimportPathapp.get(/items/{item_id})defread_item(item_id:intPath(...,description商品ID,gt0,le1000)): 获取商品详情 :param item_id: 商品ID1-1000 :return: 商品信息 return{item_id:item_id}嵌套模型Pydantic 支持嵌套模型frompydanticimportBaseModelfromtypingimportListclassAddress(BaseModel):地址模型street:strcity:strzipcode:strclassUser(BaseModel):用户模型username:stremail:straddresses:List[Address][]# 嵌套列表app.post(/users)defcreate_user(user:User):returnuser.dict()请求示例{username:john,email:johnexample.com,addresses:[{street:123 Main St,city:Beijing,zipcode:100000},{street:456 Oak Ave,city:Shanghai,zipcode:200000}]}七、异常处理HTTPException使用HTTPException抛出自定义异常fromfastapiimportHTTPException# 模拟数据库items_db{1:{name:Item 1,price:10},2:{name:Item 2,price:20}}app.get(/items/{item_id})defread_item(item_id:int): 获取商品详情 :param item_id: 商品ID :return: 商品信息 itemitems_db.get(item_id)ifnotitem:raiseHTTPException(status_code404,detailf商品{item_id}不存在,headers{X-Error:ItemNotFound})returnitem响应示例404{detail:商品 3 不存在}自定义异常处理器使用app.exception_handler自定义异常处理fromfastapiimportRequestfromfastapi.responsesimportJSONResponse# 自定义异常类classCustomException(Exception):def__init__(self,status_code:int,message:str):self.status_codestatus_code self.messagemessage# 自定义异常处理器app.exception_handler(CustomException)asyncdefcustom_exception_handler(request:Request,exc:CustomException):returnJSONResponse(status_codeexc.status_code,content{error:exc.message})app.get(/custom-error)defraise_custom_error():raiseCustomException(status_code400,message自定义错误信息)全局异常处理器处理所有未捕获的异常fromfastapiimportRequestfromfastapi.responsesimportJSONResponseimporttracebackapp.exception_handler(Exception)asyncdefglobal_exception_handler(request:Request,exc:Exception):处理所有未捕获的异常returnJSONResponse(status_code500,content{error:服务器内部错误,detail:str(exc),traceback:traceback.format_exc()})八、响应处理响应模型使用response_model指定响应格式fromfastapiimportFastAPIfrompydanticimportBaseModel appFastAPI()classUserIn(BaseModel):请求模型username:strpassword:stremail:strclassUserOut(BaseModel):响应模型username:stremail:strapp.post(/users,response_modelUserOut)defcreate_user(user:UserIn): 创建用户不返回密码 :param user: 用户信息 :return: 用户信息隐藏密码 # 实际应用中会保存到数据库returnuser请求示例{username:john,password:secret,email:johnexample.com}响应示例密码被过滤{username:john,email:johnexample.com}响应模型配置frompydanticimportBaseModelclassItem(BaseModel):name:strdescription:strNoneprice:floattax:float10.5tags:list[]items_db{foo:{name:Foo,price:50.2},bar:{name:Bar,description:The bartenders,price:62,tax:20.2}}app.get(/items/{item_id},response_modelItem)defread_item(item_id:str):returnitems_db[item_id]响应示例包含默认值{name:Foo,description:null,price:50.2,tax:10.5,tags:[]}排除未设置的字段使用response_model_exclude_unset只返回实际设置的字段app.get(/items/{item_id},response_modelItem,response_model_exclude_unsetTrue)defread_item(item_id:str):returnitems_db[item_id]响应示例不包含默认值{name:Foo,price:50.2}包含/排除特定字段app.get(/items/{item_id},response_modelItem,response_model_include{name,price})defread_item(item_id:str):returnitems_db[item_id]app.get(/items/{item_id},response_modelItem,response_model_exclude{tax})defread_item(item_id:str):returnitems_db[item_id]自定义响应状态码fromfastapiimportstatusapp.post(/items,status_codestatus.HTTP_201_CREATED)defcreate_item(name:str): 创建商品返回201状态码 :param name: 商品名称 :return: 商品信息 return{name:name}自定义响应头fromfastapiimportResponseapp.get(/custom-header)defcustom_header(response:Response): 设置自定义响应头 :param response: 响应对象 :return: 消息 response.headers[X-Custom-Header]custom-valuereturn{message:Hello}九、依赖注入基本依赖依赖注入允许我们将重复的逻辑抽取为独立的函数fromfastapiimportDepends,FastAPI appFastAPI()defget_db():获取数据库连接db数据库连接对象print(建立数据库连接)returndbapp.get(/items)defread_items(dbDepends(get_db)): 获取商品列表 :param db: 数据库连接通过依赖注入 :return: 商品列表 return{db:db,items:[]}app.get(/users)defread_users(dbDepends(get_db)): 获取用户列表 :param db: 数据库连接通过依赖注入 :return: 用户列表 return{db:db,users:[]}带 yield 的依赖资源管理使用yield实现资源的获取和释放fromfastapiimportDependsfromsqlalchemy.ormimportSessiondefget_db(): 获取数据库会话 使用 yield 确保会话正确关闭 dbSessionLocal()try:yielddbfinally:db.close()app.get(/items)defread_items(db:SessionDepends(get_db)): 获取商品列表 :param db: 数据库会话 :return: 商品列表 itemsdb.query(Item).all()returnitems多级依赖依赖可以相互依赖fromfastapiimportDependsdefget_config():获取配置return{secret_key:secret123}defget_db(configDepends(get_config)):获取数据库连接依赖配置dbf数据库连接使用密钥:{config[secret_key]}returndbapp.get(/items)defread_items(dbDepends(get_db)): 获取商品列表 :param db: 数据库连接 :return: 商品列表 return{db:db}带参数的依赖依赖函数可以接受参数fromfastapiimportDepends,Querydefpagination(skip:intQuery(0,ge0),limit:intQuery(10,ge1,le100)):分页参数依赖return{skip:skip,limit:limit}app.get(/items)defread_items(pagination_paramsDepends(pagination)): 获取商品列表分页 :param pagination_params: 分页参数 :return: 商品列表 items[{id:i,name:fItem{i}}foriinrange(100)]skippagination_params[skip]limitpagination_params[limit]returnitems[skip:skiplimit]类作为依赖使用类作为依赖fromfastapiimportDependsclassDatabase:def__init__(self):self.connection数据库连接defquery(self,sql):执行查询returnf执行SQL:{sql}defget_db():获取数据库实例dbDatabase()try:yielddbfinally:# 清理资源passapp.get(/query)defquery_db(db:DatabaseDepends(get_db)): 查询数据库 :param db: 数据库实例 :return: 查询结果 resultdb.query(SELECT * FROM users)return{result:result}全局依赖为整个应用添加全局依赖fromfastapiimportDepends,FastAPI appFastAPI(dependencies[Depends(get_db)])app.get(/items)defread_items():获取商品列表自动注入数据库依赖return{items:[]}