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

FastAPI 异常处理

news 来源：原创 2025/5/29 8:16:54

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)

测试：

相关文章：

  • 传统图像分割方法：阈值分割、Canny检测
  • 大学大模型教学：基于NC数据的全球气象可视化解决方案
  • win32相关（进程相关API）
  • 什么是DevOps的核心目标？它如何解决传统开发与运维之间的冲突？​
  • 好坏质检分类实战（异常数据检测、降维、KNN模型分类、混淆矩阵进行模型评估）
  • 7000字基于 SpringBoot 的 Cosplay 文化展示与交流社区系统设计与实现
  • Excel 统计某个字符串在指定区域出现的次数
  • 有什么excel.js支持IE11,可以显示EXCEL单元格数据,支持单元格合并,边框线,单元格背景
  • 小样本分类新突破：QPT技术详解
  • day 33 python打卡
  • SPSS跨域分类：自监督知识+软模板优化
  • SpringBoot-允许跨域配置
  • 参数/非参数检验和连续/离散/分类等变量类型的关系
  • Rust 开发的一些GUI库
  • Reactor和Proactor
  • React从基础入门到高级实战：React 核心技术 - 错误处理与错误边界：构建稳定的应用
  • React 虚拟dom
  • Rust语言学习教程、案例与项目实战指引
  • 甲醇 燃料 不也有碳排放吗？【AI回答版】
  • FPGA 的分类和发展
  • 摄影图片网站/百度搜索关键词排名查询
  • 做电影网站需要什么手续/临沂百度公司地址
  • 软件开发app开发定制外包11/淘宝seo搜索优化工具
  • 河北省建设执业注册中心网站/百度数据开放平台
  • 国产 做 视频网站/长沙网站推广智投未来
  • 淮南app/seo教程自学