FastAPI中间件自定义异常抛出后无法被处理的问题求助

我搭建了一个简单的FastAPI项目，创建了继承自BaseHTTPMiddleware的自定义中间件类。需要在该中间件的特定条件下终止执行流程，因此创建了名为CustomError的自定义异常类并抛出该异常。

from fastapi import FastAPI, Request
from starlette.middleware.base import (
    BaseHTTPMiddleware,
    RequestResponseEndpoint
)
from starlette.responses import JSONResponse, Response

app = FastAPI()


class CustomError(Exception):
    def __init__(self, message):
        self.message = message

    def __str__(self):
        return self.message


class CustomMiddleware(BaseHTTPMiddleware):
    def execute_custom_logic(self, request: Request):
        raise CustomError("This is from `CustomMiddleware`")

    async def dispatch(
            self,
            request: Request,
            call_next: RequestResponseEndpoint,
    ) -> Response:
        self.execute_custom_logic(request=request)
        response = await call_next(request)
        return response


app.add_middleware(CustomMiddleware)


@app.exception_handler(CustomError)
async def custom_exception_handler(request: Request, exc: CustomError):
    return JSONResponse(
        status_code=418,
        content={"message": exc.message},
    )


@app.get(path="/")
def root_api():
    return {"message": "Hello World"}

遗憾的是，即使添加了custom_exception_handler(...)处理器，FastAPI仍无法处理CustomError异常。

问题 #

1. FastAPI中处理此类场景的正确方式是什么？
2. 当前代码无法正常工作的原因是什么？

版本信息：

• FastAPI - 0.95.2
• Python - 3.8.13

问题2：当前代码失效的原因 #

FastAPI通过@app.exception_handler注册的异常处理器，只能捕获**路由处理函数（endpoint）**及其依赖中抛出的异常。而BaseHTTPMiddleware的dispatch方法运行在FastAPI核心异常处理栈之外，当中间件里直接抛出异常时，会直接中断请求流程，无法被FastAPI的异常处理器捕获。

问题1：正确处理方案 #

有两种实用的解决方式：

方案1：中间件内部捕获异常并返回响应

直接在dispatch方法中捕获CustomError，构造响应返回，不需要依赖全局异常处理器：

async def dispatch(
        self,
        request: Request,
        call_next: RequestResponseEndpoint,
) -> Response:
    try:
        self.execute_custom_logic(request=request)
        response = await call_next(request)
        return response
    except CustomError as exc:
        return JSONResponse(
            status_code=418,
            content={"message": exc.message},
        )

方案2：复用全局异常处理器逻辑

如果希望复用已注册的custom_exception_handler，可以在中间件捕获异常后主动调用该处理器：

async def dispatch(
        self,
        request: Request,
        call_next: RequestResponseEndpoint,
) -> Response:
    try:
        self.execute_custom_logic(request=request)
        response = await call_next(request)
        return response
    except CustomError as exc:
        # 获取已注册的异常处理器
        handler = request.app.exception_handlers.get(CustomError)
        if handler:
            return await handler(request, exc)
        # 无处理器时返回默认错误响应
        return JSONResponse(
            status_code=500,
            content={"message": "服务器内部错误"},
        )

注意：如果中间件中包含异步逻辑，要确保用await调用，避免阻塞线程。当前代码中的execute_custom_logic是同步方法，直接调用即可。

内容的提问来源于stack exchange，提问作者JPG