Перейти к содержанию

Пользовательские классы Request и APIRoute

В некоторых случаях может понадобиться переопределить логику, используемую классами Request и APIRoute.

В частности, это может быть хорошей альтернативой логике в middleware.

Например, если вы хотите прочитать или изменить тело запроса до того, как оно будет обработано вашим приложением.

Опасность

Это «продвинутая» возможность.

Если вы только начинаете работать с FastAPI, возможно, стоит пропустить этот раздел.

Сценарии использования

Некоторые сценарии:

  • Преобразование тел запросов, не в формате JSON, в JSON (например, msgpack).
  • Распаковка тел запросов, сжатых с помощью gzip.
  • Автоматическое логирование всех тел запросов.

Обработка пользовательского кодирования тела запроса

Посмотрим как использовать пользовательский подкласс Request для распаковки gzip-запросов.

И подкласс APIRoute, чтобы использовать этот пользовательский класс запроса.

Создать пользовательский класс GzipRequest

Совет

Это учебный пример, демонстрирующий принцип работы. Если вам нужна поддержка Gzip, вы можете использовать готовый GzipMiddleware.

Сначала создадим класс GzipRequest, который переопределит метод Request.body() и распакует тело запроса при наличии соответствующего HTTP-заголовка.

Если в заголовке нет gzip, он не будет пытаться распаковывать тело.

Таким образом, один и тот же класс маршрута сможет обрабатывать как gzip-сжатые, так и несжатые запросы.

import gzip
from typing import Callable, List

from fastapi import Body, FastAPI, Request, Response
from fastapi.routing import APIRoute


class GzipRequest(Request):
    async def body(self) -> bytes:
        if not hasattr(self, "_body"):
            body = await super().body()
            if "gzip" in self.headers.getlist("Content-Encoding"):
                body = gzip.decompress(body)
            self._body = body
        return self._body


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

        async def custom_route_handler(request: Request) -> Response:
            request = GzipRequest(request.scope, request.receive)
            return await original_route_handler(request)

        return custom_route_handler


app = FastAPI()
app.router.route_class = GzipRoute


@app.post("/sum")
async def sum_numbers(numbers: List[int] = Body()):
    return {"sum": sum(numbers)}

Создать пользовательский класс GzipRoute

Далее создадим пользовательский подкласс fastapi.routing.APIRoute, который будет использовать GzipRequest.

На этот раз он переопределит метод APIRoute.get_route_handler().

Этот метод возвращает функцию. Именно эта функция получает HTTP-запрос и возвращает HTTP-ответ.

Здесь мы используем её, чтобы создать GzipRequest из исходного HTTP-запроса.

import gzip
from typing import Callable, List

from fastapi import Body, FastAPI, Request, Response
from fastapi.routing import APIRoute


class GzipRequest(Request):
    async def body(self) -> bytes:
        if not hasattr(self, "_body"):
            body = await super().body()
            if "gzip" in self.headers.getlist("Content-Encoding"):
                body = gzip.decompress(body)
            self._body = body
        return self._body


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

        async def custom_route_handler(request: Request) -> Response:
            request = GzipRequest(request.scope, request.receive)
            return await original_route_handler(request)

        return custom_route_handler


app = FastAPI()
app.router.route_class = GzipRoute


@app.post("/sum")
async def sum_numbers(numbers: List[int] = Body()):
    return {"sum": sum(numbers)}

Технические детали

У Request есть атрибут request.scope — это просто Python-dict, содержащий метаданные, связанные с HTTP-запросом.

У Request также есть request.receive — функция для «получения» тела запроса.

И dict scope, и функция receive являются частью спецификации ASGI.

Именно этих двух компонентов — scope и receive — достаточно, чтобы создать новый экземпляр Request.

Чтобы узнать больше о Request, см. документацию Starlette о запросах.

Единственное, что делает по-другому функция, возвращённая GzipRequest.get_route_handler, — преобразует Request в GzipRequest.

Благодаря этому наш GzipRequest позаботится о распаковке данных (при необходимости) до передачи их в наши операции пути.

Дальше вся логика обработки остаётся прежней.

Но благодаря изменениям в GzipRequest.body тело запроса будет автоматически распаковано при необходимости, когда оно будет загружено FastAPI.

Доступ к телу запроса в обработчике исключений

Совет

Для решения этой задачи, вероятно, намного проще использовать body в пользовательском обработчике RequestValidationError (Обработка ошибок).

Но этот пример всё равно актуален и показывает, как взаимодействовать с внутренними компонентами.

Тем же подходом можно воспользоваться, чтобы получить доступ к телу запроса в обработчике исключений.

Нужно лишь обработать запрос внутри блока try/except:

from typing import Callable, List

from fastapi import Body, FastAPI, HTTPException, Request, Response
from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute


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

        async def custom_route_handler(request: Request) -> Response:
            try:
                return await original_route_handler(request)
            except RequestValidationError as exc:
                body = await request.body()
                detail = {"errors": exc.errors(), "body": body.decode()}
                raise HTTPException(status_code=422, detail=detail)

        return custom_route_handler


app = FastAPI()
app.router.route_class = ValidationErrorLoggingRoute


@app.post("/")
async def sum_numbers(numbers: List[int] = Body()):
    return sum(numbers)

Если произойдёт исключение, экземпляр Request всё ещё будет в области видимости, поэтому мы сможем прочитать тело запроса и использовать его при обработке ошибки:

from typing import Callable, List

from fastapi import Body, FastAPI, HTTPException, Request, Response
from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute


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

        async def custom_route_handler(request: Request) -> Response:
            try:
                return await original_route_handler(request)
            except RequestValidationError as exc:
                body = await request.body()
                detail = {"errors": exc.errors(), "body": body.decode()}
                raise HTTPException(status_code=422, detail=detail)

        return custom_route_handler


app = FastAPI()
app.router.route_class = ValidationErrorLoggingRoute


@app.post("/")
async def sum_numbers(numbers: List[int] = Body()):
    return sum(numbers)

Пользовательский класс APIRoute в роутере

Вы также можете задать параметр route_class у APIRouter:

import time
from typing import Callable

from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute


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

        async def custom_route_handler(request: Request) -> Response:
            before = time.time()
            response: Response = await original_route_handler(request)
            duration = time.time() - before
            response.headers["X-Response-Time"] = str(duration)
            print(f"route duration: {duration}")
            print(f"route response: {response}")
            print(f"route response headers: {response.headers}")
            return response

        return custom_route_handler


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


@app.get("/")
async def not_timed():
    return {"message": "Not timed"}


@router.get("/timed")
async def timed():
    return {"message": "It's the time of my life"}


app.include_router(router)

В этом примере операции пути, объявленные в router, будут использовать пользовательский класс TimedRoute и получат дополнительный HTTP-заголовок X-Response-Time в ответе с временем, затраченным на формирование ответа:

import time
from typing import Callable

from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute


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

        async def custom_route_handler(request: Request) -> Response:
            before = time.time()
            response: Response = await original_route_handler(request)
            duration = time.time() - before
            response.headers["X-Response-Time"] = str(duration)
            print(f"route duration: {duration}")
            print(f"route response: {response}")
            print(f"route response headers: {response.headers}")
            return response

        return custom_route_handler


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


@app.get("/")
async def not_timed():
    return {"message": "Not timed"}


@router.get("/timed")
async def timed():
    return {"message": "It's the time of my life"}


app.include_router(router)