中间件、异常处理与自定义响应 本部分目标学会捕获错误并返回友好提示使用中间件记录日志、处理跨域CORS统一 API 响应格式提升应用健壮性和用户体验。一、为什么需要异常处理和中间件用户友好默认错误信息如 422对前端不友好安全性避免暴露内部错误细节可维护性统一日志、性能监控、跨域支持一致性所有接口返回相同结构的 JSONFastAPI 提供了强大的机制来实现这些需求。二、自定义异常处理器Exception Handlers2.1 捕获内置异常FastAPI 自动处理很多异常如验证失败但你可以覆盖它。示例美化 422 错误请求数据无效# main.pyfromfastapiimportFastAPI,Requestfromfastapi.responsesimportJSONResponsefromfastapi.exceptionsimportRequestValidationError appFastAPI()app.exception_handler(RequestValidationError)asyncdefvalidation_exception_handler(request:Request,exc:RequestValidationError):returnJSONResponse(status_code400,content{error:请求参数错误,details:exc.errors()}) 现在当 JSON 字段缺失或类型错误时返回{error:请求参数错误,details:[...]}2.2 捕获自定义异常步骤1定义自定义异常类# exceptions.pyclassUserNotFoundException(Exception):def__init__(self,username:str):self.usernameusername步骤2注册异常处理器# main.pyfromexceptionsimportUserNotFoundExceptionapp.exception_handler(UserNotFoundException)asyncdefuser_not_found_handler(request:Request,exc:UserNotFoundException):returnJSONResponse(status_code404,content{error:f用户{exc.username}不存在})步骤3在路由中抛出异常app.get(/users/{username})defget_user(username:str):ifusernamenotin[alice,bob]:raiseUserNotFoundException(username)return{username:username}三、中间件Middleware中间件在每个请求前后执行通用逻辑如记录请求日志添加 CORS 头允许前端跨域测量请求耗时验证全局 Token3.1 CORS 中间件必须前端如 React/Vue通常运行在不同端口浏览器会阻止跨域请求。CORSCross-Origin Resource Sharing解决这个问题。# main.pyfromfastapi.middleware.corsimportCORSMiddleware app.add_middleware(CORSMiddleware,allow_origins[*],# 允许所有来源生产环境应限制allow_credentialsTrue,allow_methods[*],# 允许所有 HTTP 方法allow_headers[*],# 允许所有请求头)⚠️ 生产环境建议allow_origins[https://your-frontend.com]3.2 自定义中间件记录请求日志importtimefromstarlette.middleware.baseimportBaseHTTPMiddlewareclassLogMiddleware(BaseHTTPMiddleware):asyncdefdispatch(self,request:Request,call_next):start_timetime.time()responseawaitcall_next(request)process_timetime.time()-start_timeprint(f请求:{request.method}{request.url}| 耗时:{process_time:.3f}s)returnresponse app.add_middleware(LogMiddleware) 启动后每次请求会打印日志到控制台。四、统一响应格式最佳实践很多团队要求所有成功响应都遵循固定结构例如{code:200,message:success,data:{...}}实现方式自定义响应模型 封装函数# schemas.pyfrompydanticimportBaseModelfromtypingimportAny,OptionalclassResponseModel(BaseModel):code:int200message:strsuccessdata:AnyNone# 工具函数defsuccess_response(data:AnyNone,message:strsuccess):returnResponseModel(datadata,messagemessage)在路由中使用app.get(/health)defhealth_check():returnsuccess_response(data{status:OK}) 你也可以通过中间件自动包装所有响应但初学者建议先用函数封装。五、完整示例代码main.py# main.pyimporttimefromfastapiimportFastAPI,Request,HTTPExceptionfromfastapi.responsesimportJSONResponsefromfastapi.exceptionsimportRequestValidationErrorfromfastapi.middleware.corsimportCORSMiddlewarefromstarlette.middleware.baseimportBaseHTTPMiddlewarefrompydanticimportBaseModelfromtypingimportAny# 自定义异常classUserNotFoundException(Exception):def__init__(self,username:str):self.usernameusername# 统一响应模型classResponseModel(BaseModel):code:int200message:strsuccessdata:AnyNonedefsuccess_response(data:AnyNone,message:strsuccess):returnResponseModel(datadata,messagemessage)# 日志中间件classLogMiddleware(BaseHTTPMiddleware):asyncdefdispatch(self,request:Request,call_next):start_timetime.time()responseawaitcall_next(request)process_timetime.time()-start_timeprint(f⏱️{request.method}{request.url}|{process_time:.3f}s)returnresponse# 创建应用appFastAPI(title第7部分异常处理与中间件)# 添加中间件顺序很重要app.add_middleware(CORSMiddleware,allow_origins[*],allow_credentialsTrue,allow_methods[*],allow_headers[*],)app.add_middleware(LogMiddleware)# 异常处理器app.exception_handler(RequestValidationError)asyncdefvalidation_exception_handler(request:Request,exc:RequestValidationError):returnJSONResponse(status_code400,content{code:400,message:请求参数错误,data:exc.errors()})app.exception_handler(UserNotFoundException)asyncdefuser_not_found_handler(request:Request,exc:UserNotFoundException):returnJSONResponse(status_code404,content{code:404,message:f用户{exc.username}不存在,data:None})# 路由app.get(/health)defhealth_check():returnsuccess_response(data{status:OK})app.get(/users/{username})defget_user(username:str):ifusernamenotin[alice,bob]:raiseUserNotFoundException(username)returnsuccess_response(data{username:username})app.post(/items/)defcreate_item(item:dict):# 故意触发验证错误无 Pydantic 模型nameitem[name]# 如果没有 name 字段会报错returnsuccess_response(data{item:name})六、练习任务动手实践 请先自己尝试完成再查看下方答案任务1添加“文章未找到”异常定义PostNotFoundException注册异常处理器返回 404 和友好消息在/posts/{post_id}路由中使用如果文章不存在任务2限制 CORS 来源修改 CORS 配置只允许http://localhost:3000常见前端开发端口任务3挑战统一错误响应格式所有异常处理器返回{ code: ..., message: ..., data: null }确保 404、400、500 等都遵循此格式七、练习任务参考答案任务1 答案# exceptions.py新增classPostNotFoundException(Exception):def__init__(self,post_id:int):self.post_idpost_id# main.pyfromexceptionsimportPostNotFoundExceptionapp.exception_handler(PostNotFoundException)asyncdefpost_not_found_handler(request:Request,exc:PostNotFoundException):returnJSONResponse(status_code404,content{code:404,message:f文章{exc.post_id}不存在,data:None})# 在 read_post 路由中app.get(/posts/{post_id},response_modelPost)defread_post(post_id:int,session:SessionDepends(get_session)):postsession.get(Post,post_id)ifnotpost:raisePostNotFoundException(post_id)returnpost任务2 答案app.add_middleware(CORSMiddleware,allow_origins[http://localhost:3000],# ← 修改这里allow_credentialsTrue,allow_methods[*],allow_headers[*],)任务3 答案确保所有异常处理器返回统一格式# 示例HTTPException 也统一处理app.exception_handler(HTTPException)asyncdefhttp_exception_handler(request:Request,exc:HTTPException):returnJSONResponse(status_codeexc.status_code,content{code:exc.status_code,message:exc.detail,data:None}) 注意HTTPException是 FastAPI 内部使用的异常捕获它可统一 401、403 等。八、小结在本部分你学会了使用app.exception_handler自定义错误响应通过CORS 中间件解决跨域问题编写自定义中间件实现日志、性能监控设计统一响应格式提升 API 专业性