Користувацькі класи Request та APIRoute¶

🌐 Переклад ШІ та людьми

Цей переклад виконано ШІ під керівництвом людей. 🤝

Можливі помилки через неправильне розуміння початкового змісту або неприродні формулювання тощо. 🤖

Ви можете покращити цей переклад, допомігши нам краще спрямовувати AI LLM.

Англійська версія

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

Зокрема, це може бути доброю альтернативою логіці в проміжному програмному забезпеченні.

Наприклад, якщо потрібно прочитати або змінити тіло запиту до того, як його обробить ваш застосунок.

Обережно

Це «просунута» можливість.

Якщо ви тільки починаєте працювати з FastAPI, можливо, варто пропустити цей розділ.

Випадки використання¶

Деякі варіанти використання:

Перетворення не-JSON тіл запитів на JSON (наприклад, msgpack).
Розпакування тіл запитів, стиснених gzip.
Автоматичне логування всіх тіл запитів.

Обробка користувацьких кодувань тіла запиту¶

Розгляньмо, як використати користувацький підклас Request для розпакування gzip-запитів.

А також підклас APIRoute, щоб застосувати цей користувацький клас запиту.

Створіть користувацький клас `GzipRequest`¶

Порада

Це навчальний приклад, щоб продемонструвати принцип роботи. Якщо вам потрібна підтримка Gzip, скористайтеся вбудованим GzipMiddleware.

Спочатку створимо клас GzipRequest, який перевизначить метод Request.body(), щоб розпаковувати тіло за наявності відповідного заголовка.

Якщо в заголовку немає gzip, він не намагатиметься розпаковувати тіло.

Таким чином один і той самий клас маршруту зможе обробляти як стиснені gzip, так і нестиснені запити.

Python 3.10+

import gzip
from collections.abc import Callable
from typing import Annotated

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: Annotated[list[int], Body()]):
    return {"sum": sum(numbers)}

🤓 Other versions and variants

Python 3.10+ - non-Annotated

Tip

Prefer to use the Annotated version if possible.

import gzip
from collections.abc import Callable

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().

Цей метод повертає функцію. І саме ця функція прийме запит і поверне відповідь.

Тут ми використовуємо її, щоб створити GzipRequest з початкового запиту.

Python 3.10+

import gzip
from collections.abc import Callable
from typing import Annotated

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: Annotated[list[int], Body()]):
    return {"sum": sum(numbers)}

🤓 Other versions and variants

Python 3.10+ - non-Annotated

Tip

Prefer to use the Annotated version if possible.

import gzip
from collections.abc import Callable

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, що містить метадані, пов'язані із запитом.

Також Request має request.receive - це функція для «отримання» тіла запиту.

scope dict і функція receive є частиною специфікації ASGI.

І саме ці дві сутності - scope та receive - потрібні для створення нового екземпляра Request.

Щоб дізнатися більше про Request, перегляньте документацію Starlette про запити.

Єдине, що робить інакше функція, повернена GzipRequest.get_route_handler, - перетворює Request на GzipRequest.

Завдяки цьому наш GzipRequest подбає про розпакування даних (за потреби) перед передаванням їх у наші операції шляху.

Після цього вся логіка обробки залишається тією самою.

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

Доступ до тіла запиту в обробнику виключень¶

Порада

Щоб розв’язати це саме завдання, скоріш за все, простіше використати body у користувацькому обробнику RequestValidationError (Обробка помилок).

Але цей приклад усе ще корисний і показує, як взаємодіяти з внутрішніми компонентами.

Ми також можемо скористатися цим підходом, щоб отримати доступ до тіла запиту в обробнику виключень.

Усе, що потрібно, - обробити запит усередині блоку try/except:

Python 3.10+

from collections.abc import Callable
from typing import Annotated

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: Annotated[list[int], Body()]):
    return sum(numbers)

🤓 Other versions and variants

Python 3.10+ - non-Annotated

Tip

Prefer to use the Annotated version if possible.

from collections.abc import Callable

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 усе ще буде у видимості, тож ми зможемо прочитати й використати тіло запиту під час обробки помилки:

Python 3.10+

from collections.abc import Callable
from typing import Annotated

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: Annotated[list[int], Body()]):
    return sum(numbers)

🤓 Other versions and variants

Python 3.10+ - non-Annotated

Tip

Prefer to use the Annotated version if possible.

from collections.abc import Callable

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:

Python 3.10+

import time
from collections.abc 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 і матимуть додатковий заголовок відповіді X-Response-Time із часом, витраченим на формування відповіді:

Python 3.10+

import time
from collections.abc 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)

Користувацькі класи Request та APIRoute¶

Випадки використання¶

Обробка користувацьких кодувань тіла запиту¶

Створіть користувацький клас GzipRequest¶

Створіть користувацький клас GzipRoute¶

Доступ до тіла запиту в обробнику виключень¶

Користувацький клас APIRoute у маршрутизаторі¶

Створіть користувацький клас `GzipRequest`¶

Створіть користувацький клас `GzipRoute`¶

Користувацький клас `APIRoute` у маршрутизаторі¶