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

FastAPI 异常处理

backend 2025/7/28 0:08:52

HTTPException

使用 HTTPException，向客户端返回 HTTP 错误响应。

示例：

import uvicorn
from fastapi import FastAPI, HTTPExceptionapp = FastAPI()items = {"foo": "The Foo Wrestlers"}@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")return {"item": items[item_id]}if __name__ == '__main__':uvicorn.run(app)

添加自定义响应头

import uvicorn
from fastapi import FastAPI, HTTPExceptionapp = FastAPI()items = {"foo": "The Foo Wrestlers"}@app.get("/items/{item_id}")
async def read_item_header(item_id: str):if item_id not in items:raise HTTPException(status_code=404,detail="Item not found",headers={"X-Error": "There goes my error"},)if __name__ == '__main__':uvicorn.run(app)

自定义异常处理器

添加自定义处理器，要使用 Starlette 的异常工具。假设要触发的自定义异常叫作 UnicornException。且需要 FastAPI 实现全局处理该异常。

此时，可以用 @app.exception_handler() 添加自定义异常控制器：

import uvicorn
from fastapi import FastAPI,Request
from starlette.responses import JSONResponseclass UnicornException(Exception):def __init__(self, name: str):self.name = nameapp = FastAPI()@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):return JSONResponse(status_code=418,content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},)@app.get("/unicorns/{name}")
async def read_unicorn(name: str):if name == "yolo":raise UnicornException(name)return {"name": name}if __name__ == '__main__':uvicorn.run(app)

示例：

import uvicorn
from fastapi import FastAPI
from starlette.requests import Request
from starlette.responses import JSONResponseapp = FastAPI()class UserNotFoundException(Exception):"""自定义异常类-用户不存在异常"""status_code = 1001detail = "用户不存在"@app.exception_handler(UserNotFoundException)
async def user_not_found_exception_handler(request: Request, exc: UserNotFoundException):"""自定义异常处理器 - 用户不存在:param request::param exc::return:"""return JSONResponse(status_code=404,content={"code": exc.status_code, "message": exc.detail, "data": None},)user_ids = ["000","111"
]@app.get("/user/{user_id}")
async def get_user(user_id: str):# 用户不存在if user_id not in user_ids:raise UserNotFoundExceptionreturn {"code": 200, "message": "success", "data": user_id}if __name__ == '__main__':uvicorn.run(app)

覆盖默认异常处理器

FastAPI 自带了一些默认异常处理器。触发 HTTPException 或请求无效数据时，这些处理器返回默认的 JSON 响应结果。不过，也可以使用自定义处理器覆盖默认异常处理器。

覆盖请求验证异常 RequestValidationError

请求中包含无效数据时，FastAPI 内部会触发 RequestValidationError。该异常也内置了默认异常处理器。

覆盖默认异常处理器时需要导入 RequestValidationError，并用 @app.excption_handler(RequestValidationError) 装饰异常处理器。

这样，异常处理器就可以接收 Request 与异常。

import uvicorn
from fastapi import FastAPI
from fastapi.exceptions import RequestValidationError, HTTPException
from starlette.requests import Request
from starlette.responses import JSONResponseapp = FastAPI()@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc):"""覆盖请求验证异常 RequestValidationError请求中包含无效数据时，FastAPI 内部会触发 RequestValidationError:param request::param exc::return:"""return JSONResponse(status_code=400,content={"code": 400, "message": "请求参数错误", "data": None},)@app.get("/items/{item_id}")
async def read_item(item_id: int):return {"item_id": item_id}if __name__ == '__main__':uvicorn.run(app)

API接口中定义的item_id是int类型，当请求的类型是str类型时，则类型转换失败，会触发RequestValidationError。

覆盖 HTTPException 错误处理器

import uvicorn
from fastapi import FastAPI, HTTPException
from starlette.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPExceptionapp = FastAPI()@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):"""覆盖 HTTPException 错误处理器:param request::param exc::return:"""return JSONResponse(status_code=exc.status_code,content={"code": exc.status_code, "message": exc.detail, "data": None},)@app.get("/items/{item_id}")
async def read_item(item_id: int):if item_id % 2 == 0:raise HTTPException(status_code=404, detail="为查询到")return {"item_id": item_id}if __name__ == '__main__':uvicorn.run(app)

自定义全局异常

创建项目目录结构

commons包下的exceptions.py

from starlette.exceptions import HTTPException
from starlette.responses import JSONResponseasync def validation_exception_handler(request, exc):"""全局的请求参数校验异常:param request::param exc::return:"""return JSONResponse(status_code=400,content={"code": 400, "message": "请求参数错误", "data": None},)async def global_exception_handler(request, exc):"""全局的HTTP请求异常覆盖 HTTPException 错误处理器:param request::param exc::return:"""if exc.status_code == 404:code = 404message = "请求的资源不存在"elif exc.status_code == 500:code = 500message = "系统错误"else:code = exc.status_codemessage = exc.detailreturn JSONResponse(status_code=exc.status_code,content={"code": code, "message": message, "data": None},)async def buzz_exception_handler(request, exc):"""自定义业务异常处理器:param request::param exc::return:"""return JSONResponse(status_code=200,content={"code": exc.code, "message": exc.message, "data": None},)class BaseAPIException(HTTPException):"""自定义API接口异常类，继承 HTTPException模拟：当HTTP请求处理过程中遇到错误时，能够优雅地返回HTTP错误响应给客户端当API接口抛出BaseAPIException时，FastAPI会使用对应的exception_handler进行处理"""# HTTP 状态码status_code = 200# 自定义业务码code = -1# 自定义业务处理消息message = '接口服务异常'def __init__(self, message: str = None, status_code: int = None, code: int = None):self.message = message or self.messageself.status_code = status_code or self.status_codeself.code = code or self.code

创建user包

user包下的exceptions.py 用户模块的异常

from commons.exceptions import BaseAPIExceptionclass UserNotFoundException(BaseAPIException):status_code = 200code = 1001message = '用户不存在'def __init__(self, message: str = None, status_code: int = None, code: int = None):self.message = message or self.messageself.status_code = status_code or self.status_codeself.code = code or self.codeclass UserAlreadyExistsException(BaseAPIException):status_code = 200code = 1002message = '用户已经被注册了'def __init__(self, message: str = None, status_code: int = None, code: int = None):self.message = message or self.messageself.status_code = status_code or self.status_codeself.code = code or self.code

user包下的 api.py API接口

from fastapi import APIRouter, HTTPExceptionfrom user.exceptions import UserNotFoundException, UserAlreadyExistsExceptionusers = APIRouter()usernames = ["tom","jack","john",
]@users.post(path="/login", summary="用户登录")
async def login(username: str, password: str):# 用户不存在if username not in usernames:raise UserNotFoundException# raise HTTPException(status_code=400, detail="用户不存在")return {"username": username, "password": password}@users.post(path="/register", summary="用户登录")
async def register(username: str, password: str):# 用户已被注册if username in usernames:raise UserAlreadyExistsExceptionreturn {"username": username, "password": password}@users.get(path="/{username}", summary="根据用户名获取用户信息")
async def get_user_by_username(username: str):# if username not in usernames:#     raise UserNotFoundExceptionreturn {"username": username}

main.py

import uvicorn
from fastapi import FastAPI
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException
from fastapi.middleware.cors import CORSMiddlewarefrom commons.exceptions import global_exception_handler, validation_exception_handler, buzz_exception_handler, \BaseAPIException
# from my_middlewares.AuthMiddleware import AuthMiddleware
# from my_middlewares.CalculateTimeMiddleware import CalculateTimeMiddleware
from user.api import users# 异常处理器
my_exception_handlers = {# 处理HTTP请求错误HTTPException: global_exception_handler,# 处理请求参数错误RequestValidationError: validation_exception_handler,# 处理自定业务系统异常BaseAPIException: buzz_exception_handler,
}app = FastAPI(exception_handlers=my_exception_handlers)# 注册中间件
app.add_middleware(CORSMiddleware, # type: ignoreallow_origins=["*"],allow_credentials=True,allow_methods=["*"],allow_headers=["*"],
)# app.add_middleware(AuthMiddleware)  # type: ignore
# app.add_middleware(CalculateTimeMiddleware) # type: ignore# 注册路由
app.include_router(users, prefix="/users", tags=["系统用户"])if __name__ == '__main__':uvicorn.run(app)

测试：

http://www.xdnf.cn/news/9337.html

相关文章：

  • vscode ssh远程服务端设置
  • OpenCV视觉图片调整：从基础到实战的技术指南
  • PH热榜 | 2025-05-26
  • hive 笔记
  • WEB安全--RCE--webshell HIDS bypass4
  • PostgreSQL auto_explain
  • Unity3D中Mono与IL2CPP对比
  • 使用mermaid快速绘制流程图
  • 3D Tiles高级样式设置与条件渲染(3)
  • 50多种垃圾类型都能清理Wise便携版：系统临时文件 /浏览器缓存秒清理
  • 利用亮数据实现大规模数据自动抓取
  • 项目部署react经历
  • IDEA使用Git进行commit提交到本地git空间后撤回到commit版本之前
  • 本地jar包发布到maven远端
  • Vue 3.0 自定义 Composition API 管理状态
  • 银发团扎堆本地游，“微度假”模式如何盘活银发旅游市场？
  • 医疗HMI设计规范解读：如何平衡合规性与用户体验？
  • Sweet Snippet 之 指数函数优化
  • Spring AI 本地Ollama
  • 嵌入式Linux快速入门第1~2章
  • Selenium 测试框架 - Ruby
  • el-table设置自定义css
  • C语言数组遍历的方法（包含二维数组）
  • 如何构建一个高效的 iOS 应用日志体系？从开发调试到使用KeyMob上线排查的实践经验
  • vmvare 虚拟机内存不足
  • npm/yarn/pnpm安装时Sharp模块报错解决方法
  • 商品条形码查询接口如何用C#进行调用？
  • 001 flutter学习的注意事项及前期准备
  • leetcode hot100刷题日记——20.爬楼梯
  • Ubuntu实现和主机的复制粘贴 VMware-Tools（open-vm-tools）