当前位置：首页 > news >正文

FastAPI 学习教程 · 第7部分

news 2026/5/11 23:13:22

中间件、异常处理与自定义响应

💡 本部分目标：学会捕获错误并返回友好提示；使用中间件记录日志、处理跨域（CORS）；统一 API 响应格式，提升应用健壮性和用户体验。

一、为什么需要异常处理和中间件？

用户友好：默认错误信息（如 422）对前端不友好
安全性：避免暴露内部错误细节
可维护性：统一日志、性能监控、跨域支持
一致性：所有接口返回相同结构的 JSON

FastAPI 提供了强大的机制来实现这些需求。

二、自定义异常处理器（Exception Handlers）

2.1 捕获内置异常

FastAPI 自动处理很多异常（如验证失败），但你可以覆盖它。

示例：美化 422 错误（请求数据无效）

# main.pyfromfastapiimportFastAPI,Requestfromfastapi.responsesimportJSONResponsefromfastapi.exceptionsimportRequestValidationError app=FastAPI()@app.exception_handler(RequestValidationError)asyncdefvalidation_exception_handler(request:Request,exc:RequestValidationError):returnJSONResponse(status_code=400,content={"error":"请求参数错误","details":exc.errors()})

🔍 现在，当 JSON 字段缺失或类型错误时，返回：
{"error":"请求参数错误","details":[...]}

2.2 捕获自定义异常

步骤1：定义自定义异常类

# exceptions.pyclassUserNotFoundException(Exception):def__init__(self,username:str):self.username=username

步骤2：注册异常处理器

# main.pyfromexceptionsimportUserNotFoundException@app.exception_handler(UserNotFoundException)asyncdefuser_not_found_handler(request:Request,exc:UserNotFoundException):returnJSONResponse(status_code=404,content={"error":f"用户{exc.username}不存在"})

步骤3：在路由中抛出异常

@app.get("/users/{username}")defget_user(username:str):ifusernamenotin["alice","bob"]:raiseUserNotFoundException(username)return{"username":username}

三、中间件（Middleware）

中间件在每个请求前后执行通用逻辑，如：

记录请求日志
添加 CORS 头（允许前端跨域）
测量请求耗时
验证全局 Token

3.1 CORS 中间件（必须！）

前端（如 React/Vue）通常运行在不同端口，浏览器会阻止跨域请求。
CORS（Cross-Origin Resource Sharing）解决这个问题。

# main.pyfromfastapi.middleware.corsimportCORSMiddleware app.add_middleware(CORSMiddleware,allow_origins=["*"],# 允许所有来源（生产环境应限制）allow_credentials=True,allow_methods=["*"],# 允许所有 HTTP 方法allow_headers=["*"],# 允许所有请求头)

⚠️ 生产环境建议：
allow_origins=["https://your-frontend.com"]

3.2 自定义中间件：记录请求日志

importtimefromstarlette.middleware.baseimportBaseHTTPMiddlewareclassLogMiddleware(BaseHTTPMiddleware):asyncdefdispatch(self,request:Request,call_next):start_time=time.time()response=awaitcall_next(request)process_time=time.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:int=200message:str="success"data:Any=None# 工具函数defsuccess_response(data:Any=None,message:str="success"):returnResponseModel(data=data,message=message)

在路由中使用

@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.username=username# 统一响应模型classResponseModel(BaseModel):code:int=200message:str="success"data:Any=Nonedefsuccess_response(data:Any=None,message:str="success"):returnResponseModel(data=data,message=message)# 日志中间件classLogMiddleware(BaseHTTPMiddleware):asyncdefdispatch(self,request:Request,call_next):start_time=time.time()response=awaitcall_next(request)process_time=time.time()-start_timeprint(f"⏱️{request.method}{request.url}|{process_time:.3f}s")returnresponse# 创建应用app=FastAPI(title="第7部分：异常处理与中间件")# 添加中间件（顺序很重要！）app.add_middleware(CORSMiddleware,allow_origins=["*"],allow_credentials=True,allow_methods=["*"],allow_headers=["*"],)app.add_middleware(LogMiddleware)# 异常处理器@app.exception_handler(RequestValidationError)asyncdefvalidation_exception_handler(request:Request,exc:RequestValidationError):returnJSONResponse(status_code=400,content={"code":400,"message":"请求参数错误","data":exc.errors()})@app.exception_handler(UserNotFoundException)asyncdefuser_not_found_handler(request:Request,exc:UserNotFoundException):returnJSONResponse(status_code=404,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 模型）name=item["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_id=post_id# main.pyfromexceptionsimportPostNotFoundException@app.exception_handler(PostNotFoundException)asyncdefpost_not_found_handler(request:Request,exc:PostNotFoundException):returnJSONResponse(status_code=404,content={"code":404,"message":f"文章{exc.post_id}不存在","data":None})# 在 read_post 路由中@app.get("/posts/{post_id}",response_model=Post)defread_post(post_id:int,session:Session=Depends(get_session)):post=session.get(Post,post_id)ifnotpost:raisePostNotFoundException(post_id)returnpost

任务2 答案

app.add_middleware(CORSMiddleware,allow_origins=["http://localhost:3000"],# ← 修改这里allow_credentials=True,allow_methods=["*"],allow_headers=["*"],)

任务3 答案

确保所有异常处理器返回统一格式：

# 示例：HTTPException 也统一处理@app.exception_handler(HTTPException)asyncdefhttp_exception_handler(request:Request,exc:HTTPException):returnJSONResponse(status_code=exc.status_code,content={"code":exc.status_code,"message":exc.detail,"data":None})