在 Web API 开发中,优雅地处理错误和异常是保证接口质量和用户体验的关键一环。FastAPI 提供了灵活且强大的异常处理机制——既能使用内置的 HTTPException 快速返回 HTTP 错误,也能通过自定义异常处理器(Exception Handler)实现统一的错误响应格式。本文系统讲解 FastAPI 异常处理的核心知识点。
内置 HTTPException HTTPException 是 FastAPI 提供的最常用的异常类。在任何路由函数中抛出它,FastAPI 会自动返回对应的 HTTP 错误响应。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from fastapi import FastAPI, HTTPExceptionapp = FastAPI() items = {"foo" : "The Foo item" } @app.get("/items/{item_id}" ) async def read_item (item_id: str ): if item_id not in items: raise HTTPException( status_code=404 , detail="Item not found" , headers={"X-Error" : "custom-header" } ) return {"item" : items[item_id]}
访问 /items/bar 时返回:
1 2 3 { "detail" : "Item not found" }
常用参数说明:
status_code:HTTP 状态码(必须)
detail:错误详情,可以是字符串、字典或列表
headers:额外的响应头(可选)
1 2 3 4 5 6 7 8 9 10 11 12 13 14 raise HTTPException( status_code=422 , detail={"error" : "Invalid input" , "field" : "username" } ) raise HTTPException( status_code=400 , detail=[ {"loc" : ["body" , "name" ], "msg" : "name is required" }, {"loc" : ["body" , "age" ], "msg" : "age must be positive" } ] )
自定义异常处理器 通过 @app.exception_handler() 装饰器,可以为特定异常类型注册自定义处理器,实现统一的错误响应格式。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 from fastapi import FastAPI, Requestfrom fastapi.responses import JSONResponsefrom fastapi.exceptions import HTTPExceptionapp = FastAPI() class UserNotFoundException (Exception ): def __init__ (self, user_id: int ): self .user_id = user_id self .message = f"User {user_id} not found" class PermissionDeniedException (Exception ): def __init__ (self, message: str ): self .message = message @app.exception_handler(UserNotFoundException ) async def user_not_found_handler (request: Request, exc: UserNotFoundException ): return JSONResponse( status_code=404 , content={ "code" : 404 , "message" : exc.message, "data" : None } ) @app.exception_handler(HTTPException ) async def http_exception_handler (request: Request, exc: HTTPException ): return JSONResponse( status_code=exc.status_code, content={ "code" : exc.status_code, "message" : exc.detail, "data" : None } ) @app.get("/users/{user_id}" ) async def get_user (user_id: int ): if user_id > 100 : raise UserNotFoundException(user_id=user_id) return {"id" : user_id, "name" : "Alice" }
访问 /users/999 时返回:
1 2 3 4 5 { "code" : 404 , "message" : "User 999 not found" , "data" : null }
处理请求验证错误 当请求数据不符合 Pydantic 模型定义时,FastAPI 会抛出 RequestValidationError。可以自定义处理器来统一格式:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 from fastapi.exceptions import RequestValidationErrorfrom fastapi.encoders import jsonable_encoder@app.exception_handler(RequestValidationError ) async def validation_exception_handler (request: Request, exc: RequestValidationError ): errors = [] for error in exc.errors(): errors.append({ "field" : "." .join(str (loc) for loc in error["loc" ]), "message" : error["msg" ] }) return JSONResponse( status_code=422 , content={ "code" : 422 , "message" : "输入参数验证失败" , "errors" : errors } ) from pydantic import BaseModel, Fieldclass CreateUserRequest (BaseModel ): name: str = Field(..., min_length=2 , max_length=50 ) email: str = Field(..., pattern=r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$" ) age: int = Field(..., ge=0 , le=150 ) @app.post("/users" ) async def create_user (user: CreateUserRequest ): return {"user" : user}
发送非法数据时返回:
1 2 3 4 5 6 7 8 { "code" : 422 , "message" : "输入参数验证失败" , "errors" : [ { "field" : "body.email" , "message" : "string does not match regex" } , { "field" : "body.age" , "message" : "ensure this value is greater than or equal to 0" } ] }
全局捕获未处理异常 对于未预见的异常(如代码逻辑错误),注册一个兜底处理器,避免返回 500 时暴露内部细节:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 import tracebackimport logginglogger = logging.getLogger(__name__) @app.exception_handler(Exception ) async def global_exception_handler (request: Request, exc: Exception ): logger.error(f"Unhandled error: {traceback.format_exc()} " ) return JSONResponse( status_code=500 , content={ "code" : 500 , "message" : "服务器内部错误,请稍后重试" , "data" : None } )
注意 :如果同时注册了 HTTPException 和 Exception 的处理器,FastAPI 会匹配最具体的异常类型,即 HTTPException 由对应的处理器捕获,其他未处理的异常才进入全局 Exception 处理器。
使用 add_exception_handler 除了装饰器方式,还可以通过 add_exception_handler() 方法动态注册:
1 2 3 4 from starlette.exceptions import HTTPException as StarletteHTTPExceptionapp.add_exception_handler(StarletteHTTPException, http_exception_handler)
这在你想用同一个处理器处理多种异常,或在路由模块化拆分时需要在 APIRouter 之外集中管理异常处理时特别有用。
实际项目中的统一响应封装 在实际项目中,通常会将自定义异常、处理器和统一响应模型封装在一起:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 class AppException (Exception ): """应用层基础异常""" def __init__ (self, code: int , message: str ): self .code = code self .message = message class NotFoundError (AppException ): def __init__ (self, message: str = "资源未找到" ): super ().__init__(code=404 , message=message) class AuthenticationError (AppException ): def __init__ (self, message: str = "认证失败" ): super ().__init__(code=401 , message=message) class ForbiddenError (AppException ): def __init__ (self, message: str = "权限不足" ): super ().__init__(code=403 , message=message) from fastapi import FastAPI, Requestfrom fastapi.responses import JSONResponsedef register_exception_handlers (app: FastAPI ): @app.exception_handler(AppException ) async def app_exception_handler (request: Request, exc: AppException ): return JSONResponse( status_code=exc.code, content={ "code" : exc.code, "message" : exc.message, "data" : None } )
这样在任意路由中只需 raise NotFoundError("用户不存在") 就能返回统一格式的响应,整个项目的错误处理风格保持一致。
异常处理执行顺序 当抛出异常时,FastAPI 按以下优先级匹配处理器:
自定义业务异常处理器 (最具体)
HTTPException 处理器
**RequestValidationError / WebSocketRequestValidationError**(Starlette 内建异常)
兜底 Exception 处理器 (最宽泛)
如果某个异常没有被任何处理器捕获,Starlette 会返回默认的 500 纯文本响应。
小结
场景
推荐方式
简单 404/401/403 错误
直接 raise HTTPException
请求参数校验失败
处理 RequestValidationError
业务层错误
定义自定义异常 + 对应处理器
未知错误兜底
全局 Exception 处理器
统一响应格式
封装基础异常类 + 集中注册
掌握异常处理机制后,你的 FastAPI 项目就能为客户端提供清晰、一致、可靠的错误反馈——这是优秀 API 不可或缺的品质。