Python Fastapi实现统一处理各种异常

Written by

in

文章目录
  • from fastapi import FastAPI, Request, status from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse app = FastAPI() @app.exception_handler(RequestValidationError) async def validation_exception_handler(request: Request, exc: RequestValidationError): return JSONResponse( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, content={ “code”: 422, “message”: “参数校验失败”, “errors”: exc.errors() }, ) # 示例路由 @app.get(“/items/{item_id}”) async def read_item(item_id: int, q: str): return {“item_id”: item_id, “q”: q}
  • @app.exception_handler(RequestValidationError) async def custom_validation_handler(request: Request, exc: RequestValidationError): simplified_errors = [] for error in exc.errors(): # 提取关键信息 simplified_errors.append({ “field”: “->”.join(str(loc) for loc in error[“loc”]), # 错误字段路径 “msg”: error[“msg”], “type”: error[“type”] }) return JSONResponse( status_code=422, content={ “error”: “ValidationError”, “detail”: simplified_errors } )
  • from pydantic import BaseModel, Field class Item(BaseModel): name: str = Field(…, min_length=3) price: float = Field(gt=0) @app.post(“/items/”) async def create_item(item: Item): return item # 测试无效请求 # curl -X POST ‘http://localhost:8000/items/’ # -H ‘Content-Type: application/json’ # -d ‘{“name”: “AB”, “price”: -1}’
  • 如果需要统一处理其他错误: from fastapi import HTTPException @app.exception_handler(HTTPException) async def http_exception_handler(request: Request, exc: HTTPException): return JSONResponse( status_code=exc.status_code, content={“error”: exc.detail} )
  • 目录
    • 1. 基础实现(返回标准错误结构)
    • 2. 自定义错误格式(简化输出)
    • 3. 完整示例(含路由演示)
    • 4. 处理自定义错误类型(扩展)
      • 关键说明
      • 测试验证

    在 FastAPI 中处理参数校验错误(通常由 Pydantic 模型或路径参数校验触发)需要使用自定义异常处理器捕获 RequestValidationError。以下是详细实现步骤:

    from fastapi import FastAPI, Request, status
    from fastapi.exceptions import RequestValidationError
    from fastapi.responses import JSONResponse
    
    app = FastAPI()
    
    @app.exception_handler(RequestValidationError)
    async def validation_exception_handler(request: Request, exc: RequestValidationError):
        return JSONResponse(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            content={
                "code": 422,
                "message": "参数校验失败",
                "errors": exc.errors()
            },
        )
    
    # 示例路由
    @app.get("/items/{item_id}")
    async def read_item(item_id: int, q: str):
        return {"item_id": item_id, "q": q}
    

    @app.exception_handler(RequestValidationError)
    async def custom_validation_handler(request: Request, exc: RequestValidationError):
        simplified_errors = []
        for error in exc.errors():
            # 提取关键信息
            simplified_errors.append({
                "field": "->".join(str(loc) for loc in error["loc"]),  # 错误字段路径
                "msg": error["msg"],
                "type": error["type"]
            })
        
        return JSONResponse(
            status_code=422,
            content={
                "error": "ValidationError",
                "detail": simplified_errors
            }
        )
    

    from pydantic import BaseModel, Field
    
    class Item(BaseModel):
        name: str = Field(..., min_length=3)
        price: float = Field(gt=0)
    
    @app.post("/items/")
    async def create_item(item: Item):
        return item
    
    # 测试无效请求
    # curl -X POST 'http://localhost:8000/items/' 
    # -H 'Content-Type: application/json' 
    # -d '{"name": "AB", "price": -1}'
    

    如果需要统一处理其他错误:

    from fastapi import HTTPException
    
    @app.exception_handler(HTTPException)
    async def http_exception_handler(request: Request, exc: HTTPException):
        return JSONResponse(
            status_code=exc.status_code,
            content={"error": exc.detail}
        )
    

    1.错误信息结构

    {
      "loc": ["body", "price"],
      "msg": "ensure this value is greater than 0",
      "type": "value_error.number.not_gt"
    }
    
    • loc:错误位置(body/query/path/header等 + 字段路径)
    • msg:人类可读错误描述
    • type:错误类型标识符

    2.HTTP 状态码

    参数校验错误通常返回 422 Unprocessable Entity

    路径参数错误(如 item_id: int 接收到字符串)会触发相同错误

    3.调试信息: 生产环境中建议隐藏详细错误,可通过环境变量控制:

    import os
    
    @app.exception_handler(RequestValidationError)
    async def handler(...):
        if os.getenv("ENV") == "prod":
            return JSONResponse(status_code=422, content={"error": "Invalid params"})
        else:
            # 返回完整错误
    

    使用无效参数请求示例路由,将得到类似响应:

    {
      "code": 422,
      "message": "参数校验失败",
      "errors": [
        {
          "field": "body->name",
          "msg": "ensure this value has at least 3 characters",
          "type": "value_error.any_str.min_length"
        },
        {
          "field": "body->price",
          "msg": "ensure this value is greater than 0",
          "type": "value_error.number.not_gt"
        }
      ]
    }
    

    提示:FastAPI 自动生成的文档(Swagger UI)中的参数校验错误由内置处理器处理,自定义处理器不影响文档界面行为。

    到此这篇关于Python Fastapi实现统一处理各种异常的文章就介绍到这了,更多相关Python Fastapi异常处理内容请搜索风君子博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持风君子博客!

    您可能感兴趣的文章:

    • Python文件操作基础及异常处理
    • 在Python中执行异常处理的基本步骤
    • 如何在Python中自定义异常类与异常处理机制
    • Python异常处理与日志记录的操作过程
    • Python中异常处理及最佳实践举例详解
    • Python中的异常处理以及自定义异常类型方式

    站内搜索