Python中内置了完整的Exception处理体系,包括抛出异常的raise语句、捕获并处理异常的try-except-finally语句以及派生自BaseException的内置的和自定义的异常类。但为了在FastAPI应用中友好地向endpoint用户传递异常信息,FastAPI使用HTTPException类、exception_handler装饰器以及middlerware中间件对用户代码中的Exception进行包装呈现。

使用HTTPException

HTTPException是一个派生自BaseException的异常类,但它同时可以把异常信息作为Http Response返回。这样当我们在router中,甚至在简单的业务代码中也可以raise一个HTTPException或者是派生至HTTPException的异常,从而向endpoint用户传递友好的异常信息。
1、如果不使用HTTPException那么客户端只能得到粗略不友好的异常信息

import uvicorn as uvicorn
from fastapi import FastAPI, HTTPException

app = FastAPI()

# 业务逻辑
def do_some_business_logic():
    raise Exception("I do not want to do any work!")


@app.get("/")
async def index():
    do_some_business_logic()
    return "test exception"

在这里插入图片描述
2、使用了HTTPException,可以返回友好的异常信息,甚至可以自定义Response header

import http
import uvicorn as uvicorn
from fastapi import FastAPI, HTTPException

app = FastAPI()

# 业务逻辑
def do_some_business_logic():
    raise Exception("I do not want to do any work!")


@app.get("/")
async def index():
    try:
        do_some_business_logic()
    except Exception as e:
        raise HTTPException(
            status_code=http.HTTPStatus.INTERNAL_SERVER_ERROR,
            # detail 可以传递任何可以转换成JSON格式的数据
            detail=str(e),
            # 自定义新的response header
            headers={"X-Error": f'worker said:{str(e)}'},
        )
    return "test exception"

在这里插入图片描述

使用exception_handler统一处理FastAPI异常

使用HTTPException基本可以向endpoint用户传递一个友好的HTTP异常,但是我们要传递很多,怎么办呢?在每一个router中都写上try-except,默默地把来自其他地方的异常包装成HTTPException?这样也太不程序猿了。因此我们需要exception_handler装饰器,它可以一劳永逸。

import http
import uvicorn as uvicorn
from fastapi import FastAPI, HTTPException
from starlette.requests import Request
from starlette.responses import JSONResponse

app = FastAPI()


def do_some_business_logic():
    raise Exception("I do not want to do any work!")


@app.exception_handler(Exception)
async def general_exception_handler(request: Request, e: Exception):
    return JSONResponse(
        status_code=http.HTTPStatus.INTERNAL_SERVER_ERROR,
        content={"message": str(e)},
    )

@app.get("/")
async def index():
    do_some_business_logic()
    return "test exception"

在这里插入图片描述

当然也可以自定义一个Exception,并且为这个Exception指定一个exception_handler,以满足显示更多信息,记录错误日志等更加细节的需求。

import http
import uvicorn as uvicorn
from fastapi import FastAPI, HTTPException
from starlette.requests import Request
from starlette.responses import JSONResponse

app = FastAPI()

# 自定义业务逻辑异常
class BusinessException(Exception):
    def __init__(self, name: str, message: str):
        self.message = message
        self.name = name

# 业务逻辑
def do_some_business_logic(worker_name: str):
    raise BusinessException(worker_name, "I do not want to do any work!")

# 为BusinessException异常定义的处理函数(可以为任何自定义异常、内置异常定义独特的处理函数)
@app.exception_handler(BusinessException)
async def business_exception_handler(request: Request, e: BusinessException):
    return JSONResponse(
        status_code=http.HTTPStatus.INTERNAL_SERVER_ERROR,
        content={"name": e.name, "message": e.message},
    )

# 为一般异常定义的处理函数
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, e: Exception):
    return JSONResponse(
        status_code=http.HTTPStatus.INTERNAL_SERVER_ERROR,
        content={"message": str(e)},
    )

@app.get("/")
async def index():
    do_some_business_logic('tom')
    return "test exception"

在这里插入图片描述

使用exception middlerware

如果我们真的针对不同类型的Exception定义了大量的exception_handler时,那么我们main.py就会显得非常臃肿,这时我们可以使用starlette提供的ExceptionMiddleware将大量的exception_handler引入到程序中。
main.py

import uvicorn as uvicorn
from fastapi import FastAPI
from starlette.middleware.exceptions import ExceptionMiddleware
from fapi.exception_handlers import general_exception_handler

app = FastAPI()

app.add_middleware(ExceptionMiddleware, handlers={Exception: general_exception_handler})


def do_some_business_logic():
    raise Exception("I do not want to do any work!")


@app.get("/")
async def index():
    do_some_business_logic()
    return "test exception"

exception_handlers.py

import http
from starlette.requests import Request
from starlette.responses import JSONResponse


async def general_exception_handler(request: Request, e: Exception):
    return JSONResponse(
        status_code=http.HTTPStatus.INTERNAL_SERVER_ERROR,
        content={"message": str(e)},
    )

综上,FastAPI所包含的Exception处理体系基本可以满足大、中、小规模项目的需要,我们可以根据自己项目的大小来选择合适的应用方式。

Logo

开放原子开发者工作坊旨在鼓励更多人参与开源活动,与志同道合的开发者们相互交流开发经验、分享开发心得、获取前沿技术趋势。工作坊有多种形式的开发者活动,如meetup、训练营等,主打技术交流,干货满满,真诚地邀请各位开发者共同参与!

更多推荐