FastAPI中间件抛出的异常无法被自定义异常处理器处理

我想配置一个能处理应用内所有异常的全局异常处理器，目前中间件可以正常抛出InvalidAuthorizationTokenError等自定义异常，但这些异常无法被已注册的自定义异常处理器捕获处理。尝试过添加独立的ExceptionMiddleware、将异常与处理器转为字典的方案，问题仍未解决。

相关代码 #

main.py #

from typing import Any, Optional
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from dependency_injector import containers, providers
import logging.config

LOGGING_CONFIG = {}  # 省略具体配置
DB_URL = "your-db-url"
DB_MAX_CONNECTION_COUNT = 10

class NewFastApi(FastAPI):
    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        self.container: Optional[containers.Container] = None


def setup_exception_handlers(app: NewFastApi) -> NewFastApi:
    from .handlers import (
        http400_invalid_client_handler,
        http400_invalid_grant_handler,
        http400_unsupported_grant_type_handler,
        http401_invalid_token_handler
    )
    from .exceptions import (
        InvalidClientIdError,
        InvalidGrantError,
        InvalidRedirectUriError,
        UnsupportedGrantTypeError,
        InvalidAuthorizationTokenError
    )
    
    app.add_exception_handler(InvalidClientIdError, http400_invalid_client_handler)
    app.add_exception_handler(InvalidGrantError, http400_invalid_grant_handler)
    app.add_exception_handler(InvalidRedirectUriError, http400_invalid_grant_handler)
    app.add_exception_handler(UnsupportedGrantTypeError, http400_unsupported_grant_type_handler)
    app.add_exception_handler(InvalidAuthorizationTokenError, http401_invalid_token_handler)
    return app

def setup_di(app: FastAPI) -> None:
    import prov
    import ui
    
    db_engine = prov.provide_db(
        database_url=DB_URL, max_connection_count=DB_MAX_CONNECTION_COUNT
    )

    app.add_middleware(
        middleware_class=AuthorizationMiddleware,
        blacklisted_repo=prov.provide_blacklisted_repo(db_engine),
    )
    app.add_middleware(
        middleware_class=AccessTokenMiddleware,
        blacklisted_repo=prov.provide_blacklisted_repo(db_engine),
    )
    
    # 注册admin-ui控制器
    admin = ui.CustomAdmin(
        app,
        db_engine,
        templates_dir="templates_admin_ui",
        authentication_backend=ui.AdminAuthController(
            secret_key="1111",
            auth_service=prov.provide_admin_auth_service(
                user_repo=prov.provide_user_repo(db_engine),
                password_service=prov.provide_password_service(),
                jwt_service=prov.provide_jwt_service(),
            ),
        ),
    )

def get_application(test: bool = False) -> NewFastApi:
    # 配置日志
    logging.config.dictConfig(LOGGING_CONFIG)

    application = NewFastApi()

    # 先注册异常处理器，再添加中间件
    application = setup_exception_handlers(application)
    setup_di(application)

    application.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    container = containers.Container()
    container.db()
    application.container = container

    # 导入路由
    from .routers import router
    application.include_router(router)
    application.mount(
        "/static",
        StaticFiles(directory="src/presentation/api/templates/static"),
        name="static",
    )

    return application

app = get_application()

middleware.py #

from typing import Any, Callable
from fastapi import Request
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.status import HTTP_401_UNAUTHORIZED
from .exceptions import InvalidAuthorizationTokenError
from .services import JWTService
from .repositories import BlacklistedTokenRepository

REQUESTS_WITH_AUTH = [
    {"method": "GET", "path": "/userinfo/"},
    {"method": "GET", "path": "/userinfo/jwt"},
    {"method": "POST", "path": "/userinfo/"},
    {"method": "POST", "path": "/introspection/"},
    {"method": "POST", "path": "/revoke/"},
]

class AuthorizationMiddleware(BaseHTTPMiddleware):
    def __init__(
        self,
        app: Any,
        blacklisted_repo: BlacklistedTokenRepository,
        jwt_service: JWTService = JWTService(),
    ) -> None:
        super().__init__(app)  # 调用父类初始化
        self.jwt_service = jwt_service
        self.blacklisted_repo = blacklisted_repo

    async def dispatch(
        self, request: Request, call_next: Callable[..., Any]
    ) -> Any:
        for request_with_auth in REQUESTS_WITH_AUTH:
            if (
                request_with_auth["path"] == request.url.path
                and request_with_auth["method"] == request.method
            ):
                token = request.headers.get("authorization") or request.headers.get(
                    "auth-swagger"
                )
                if token is None:
                    # 手动调用应用的异常处理器
                    handler = self.app.exception_handlers.get(InvalidAuthorizationTokenError)
                    if handler:
                        return await handler(request, InvalidAuthorizationTokenError())
                    # fallback到默认响应
                    headers = {"Cache-Control": "no-store", "Pragma": "no-cache"}
                    content = {
                        "error": "invalid_token",
                        "error_description": "Incorrect authorization token",
                    }
                    return JSONResponse(
                        content=content, headers=headers, status_code=HTTP_401_UNAUTHORIZED
                    )
        # 必须调用call_next让请求继续处理
        return await call_next(request)

http401_invalid_token.py #

from fastapi import Request
from fastapi.responses import JSONResponse
from starlette.status import HTTP_401_UNAUTHORIZED
from .exceptions import InvalidAuthorizationTokenError

async def http401_invalid_token_handler(
    _: Request, exc: InvalidAuthorizationTokenError
) -> JSONResponse:
    headers = {"Cache-Control": "no-store", "Pragma": "no-cache"}
    content = {
        "error": "invalid_token",
        "error_description": "Incorrect authorization token",
    }
    return JSONResponse(
        content=content, headers=headers, status_code=HTTP_401_UNAUTHORIZED
    )

解决方案 #

问题根源 #

FastAPI的BaseHTTPMiddleware基于Starlette实现，中间件的执行逻辑在应用的异常处理器之前。直接在中间件中抛出异常时，不会触发已注册的全局异常处理器，而是返回Starlette默认的错误响应。另外，原中间件的dispatch_func（应为dispatch方法）缺少return await call_next(request)，会导致非认证请求无法正常处理。

修复步骤 #

1. 修复中间件请求流转逻辑：确保匹配完认证规则后，必须调用call_next(request)让请求继续向后处理，避免非认证请求被卡住。

2. 手动调用全局异常处理器：在中间件中需要抛出自定义异常时，直接从应用的exception_handlers字典中获取对应的处理器，调用并返回其生成的响应，复用已注册的异常处理逻辑。

3. 可选：改用依赖注入实现认证：如果希望异常自动被全局处理器捕获，推荐将认证逻辑封装为FastAPI依赖（Dependency）而非中间件。依赖中抛出的异常会自动触发对应的全局异常处理器，示例如下：

# 认证依赖
from fastapi import Depends, Request
from .exceptions import InvalidAuthorizationTokenError

async def require_auth(request: Request):
    token = request.headers.get("authorization") or request.headers.get("auth-swagger")
    if not token:
        raise InvalidAuthorizationTokenError
    # 其他认证逻辑
    return token

# 在路由中使用
@router.get("/userinfo/")
async def get_userinfo(token: str = Depends(require_auth)):
    # 业务逻辑
    pass

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