在使用 FastAPI 开发接口时我们通常很快就能完成基本的 CRUD 功能。但随着项目复杂度提升仅仅会写接口是不够的还需要处理一些工程化问题比如异常统一处理、请求拦截、公共依赖管理以及接口测试。这些能力主要依赖 FastAPI 的几个核心机制异常处理Exception Handling中间件Middleware依赖注入Dependency Injection掌握这些内容可以让你的 FastAPI 项目从“能用”提升到“规范、可维护”。 接下来我们逐个来看。全局统一返回结构Result在实际项目中为了让前后端交互更规范通常会统一接口返回格式比如Result(code,msg,data){code:200,msg:success,data:{}}定义统一返回对象可以先封装一个通用结构fromtypingimportAny,Generic,TypeVarfrompydanticimportBaseModel TTypeVar(T)classResult(BaseModel,Generic[T]):code:int200msg:strsuccessdata:T|NoneNoneclassmethoddefok(cls,data:T|NoneNone,msg:strsuccess,code:int200)-Result[T]:returncls(codecode,msgmsg,datadata)classmethoddeferror(cls,code:int,msg:str,data:AnyNone)-Result[Any]:returncls(codecode,msgmsg,datadata)接口中使用app.get(/user)defget_user():returnResult.success(data{name:Tom})异常处理Exception HandlingFastAPI 的异常处理分两层内置异常HTTPException这是最常用的router.get(/detail/{user_id})asyncdefget_user_detail(user_id:int):# 路径参数会根据类型注解自动转换ifuser_id!1:raiseHTTPException(status_code404,detailUser not found)return{user_id:user_id}当我们访问user_id为2的用户就会发生异常自定义异常 全局处理你可以定义自己的异常classMyException(Exception):def__init__(self,msg:str):self.msgmsg然后注册全局异常处理器fromfastapiimportRequestfromfastapi.responsesimportJSONResponse# app FastAPI()app.exception_handler(MyException)asyncdefmy_exception_handler(request:Request,exc:MyException):returnJSONResponse(contentResult.error(code500,msgexc.msg).model_dump())注意点app.exception_handler(...)是全局生效只要是这个 app 下的所有路由都会生效修改接口router.get(/detail/{user_id})asyncdefget_user_detail(user_id:int):ifuser_id!1:raiseMyException(用户id异常)return{user_id:user_id}进行测试Pydantic 校验异常也统一pydantic校验不通过会抛出RequestValidationError我们对RequestValidationError也进行处理。注册异常处理器app.exception_handler(RequestValidationError)asyncdefvalidation_exception_handler(request:Request,exc:RequestValidationError):returnJSONResponse(contentResult.error(code422,msg参数校验失败,dataexc.errors()).model_dump())用之前的接口测试classUserCreateRequest(BaseModel):# 定义请求体字段包含参数校验user_name:strField(min_length2,max_length20)age:intField(ge0,le150)router.post()asyncdefcreate_user(request:UserCreateRequest):returnrequest.model_dump()测试一下返回结果中间件Middleware中间件本质是请求进来 → 处理 → 响应出去 的“拦截器”执行顺序请求 → middleware1 → middleware2 → 路由 → middleware2 → middleware1 → 返回能干嘛记录日志log鉴权token校验请求耗时统计CORS跨域Trace ID链路追踪最基础中间件fromfastapiimportRequest# http 不是名字而是“中间件类型app.middleware(http)asyncdeflog_middleware(request:Request,call_next):print(请求来了:,request.url)responseawaitcall_next(request)print(响应返回)returnresponseapp.middleware(http)asyncdeflog_middleware2(request:Request,call_next):print(请求来了2:,request.url)responseawaitcall_next(request)print(响应返回2)returnresponse调用一下http://localhost:8000/user/detail/2接口并查看打印日志请求来了2: http://localhost:8000/user/detail/2 请求来了: http://localhost:8000/user/detail/2 响应返回 响应返回2这还有一个美名叫洋葱模型log_middleware2 ← 外层最后定义 log_middleware route log_middleware log_middleware2 ← 外层返回依赖注入在 FastAPI 中依赖注入Depends就是把函数执行结果自动注入到参数里 自动传给接口而不是在接口里手动创建。依赖注入的常见用途获取数据库连接database.pyfromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportsessionmaker DATABASE_URLsqlite:///./test.dbenginecreate_engine(DATABASE_URL,connect_args{check_same_thread:False})SessionLocalsessionmaker(autocommitFalse,autoflushFalse,bindengine)依赖获取 DB Sessionfrom.databaseimportSessionLocaldefget_db():dbSessionLocal()try:yielddbfinally:db.close()使用 DBfromfastapiimportDependsfromsqlalchemy.ormimportSessionapp.get(/users)defget_users(db:SessionDepends(get_db)):usersdb.query(User).all()returnusers获取当前用户模拟 token 获取fromfastapiimportHeader,HTTPExceptiondefget_token(authorization:strHeader(None)):ifnotauthorization:raiseHTTPException(status_code401,detailNo token)# 简化Bearer xxxtokenauthorization.replace(Bearer ,)returntoken模拟解析用户# 这里依赖注入了get_tokendefget_current_user(token:strDepends(get_token)):# 模拟 token 解析iftoken!valid-token:raiseHTTPException(status_code401,detailInvalid token)return{id:1,username:Tom}嵌套依赖在 FastAPI 中依赖注入不仅可以用在接口函数上还可以在依赖函数里继续依赖其他依赖这就是嵌套依赖。示例伪代码最底层依赖DBdefget_db():dbdb_sessiontry:yielddbfinally:print(关闭 DB)中间层依赖用户依赖 DBfromfastapiimportDependsdefget_current_user(dbDepends(get_db)):return{id:1,username:Tom,db:db}接口层依赖用户app.get(/profile)defprofile(userDepends(get_current_user)):returnuser执行流程当请求/profile时执行顺序是profile ↓ get_current_user ↓ get_db ↓ 返回 db ↓ 返回 user ↓ 返回 response关于 FastAPI 的一些进阶功能异常处理、中间件、依赖注入。通过这些内容我们已经可以让一个 FastAPI 项目具备基本的工程化能力。也希望对大家有所帮助。下一期我们再继续深入聊一聊 FastAPI 项目的架构设计比如如何分层、如何组织模块以及如何在实际项目中落地。