Как вызвать пользовательские исключения в промежуточном программном обеспечении FastAPI?

Пользовательские обработчики исключений FastAPI не обрабатывают исключения промежуточного уровня. Хотя это нигде не указано в документах, есть часть об HTTPException , в которой говорится, что вы можете поднять HTTPException, если вы находитесь внутри служебной функции, которую вы вызываете внутри своей функции операции пути. HTTPException имеет обработчик исключений по умолчанию, который действует абсолютно так же, как и пользовательские обработчики исключений.

Вы можете либо обработать свою ошибку (с помощью try/except) в одном и том же промежуточном программном обеспечении, либо иметь отдельное промежуточное программное обеспечение, например. ExceptionHandlerMiddleware (но вам придется соблюдать правильный порядок цепочки промежуточного программного обеспечения).

Очевидным способом было бы вызвать HTTPException; однако в промежуточном программном обеспечении FastAPI/Starlette это не сработает, что приведет к ошибке Exception in ASGI application на стороне сервера, и, следовательно, клиенту будет возвращен Internal Server Error.

Вариант 1 - Использование блока middleware и try/except

Вы можете использовать блок try/except для обработки пользовательского исключения, возникшего в вашей пользовательской функции. После возникновения ошибки вы можете вернуть JSONResponse (или пользовательский Response, если хотите), включая msg (и любые другие аргументы) из CustomException, а также желаемый status_code (в примере, приведенном ниже, код состояния 500 используется, который можно заменить кодом состояния по вашему выбору).

Рабочий пример

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()


class CustomException(Exception):
    def __init__(self, msg: str):
        self.msg = msg
 
 
def exec_custom_logic(request: Request):
    raise CustomException(msg='Something went wrong') 

    
@app.middleware("http")
async def custom_middleware(request: Request, call_next):
    try:    
        exec_custom_logic(request)
    except CustomException as e:
        return JSONResponse(status_code=500, content = {'message': e.msg})
        
    return await call_next(request)
    
    
@app.get('/')
async def main(request: Request):
    return 'OK'

Вариант 2. Использование APIRouter с пользовательским классом APIRoute

Вы можете использовать APIRouter с пользовательским классом APIRoute , как показано в варианте 4 этого ответа, и либо обработать пользовательское исключение внутри блока try/except (как показано в предыдущем варианте выше), либо поднимите HTTPException напрямую. Преимущества такого подхода: (1) вы можете поднять HTTPException напрямую, и, следовательно, нет необходимости использовать блоки try/except, и (2) вы можете добавить в APIRouter только те маршруты, которые вы хотели бы обрабатывать таким образом. , используя, например, декоратор @router.get(), а остальные маршруты можно добавить в экземпляр app, используя, например, декоратор @app.get().

Рабочий пример

from fastapi import FastAPI, APIRouter, Response, Request, HTTPException
from fastapi.routing import APIRoute
from typing import Callable


def exec_custom_logic(request: Request):
    raise HTTPException(status_code=500, detail='Something went wrong')
    

class CustomAPIRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            exec_custom_logic(request)
            return await original_route_handler(request)
       
        return custom_route_handler


app = FastAPI()
router = APIRouter(route_class=CustomAPIRoute)


@router.get('/')
async def main(request: Request):
    return 'OK'
    
app.include_router(router)