Binary

Чтение тела запроса в виде последовательности байт.

Описание

Binary (MIME-type: application/octet-stream) — двоичный тип данных.

Rapidy позволяет извлекать любые данные с content_type в виде последовательности байт.

Просто укажите аннотацию bytes или StreamReader.

Зачем это нужно?

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

Есть только два типа данных, которые можно извлекать без учета content_type: bytes и StreamReader.

---

bytes

from rapidy.http import post, Body, ContentType

@post('/')
async def handler(
    user_data: bytes = Body(),
    # or use any content_type
    user_data: bytes = Body(content_type=ContentType.stream),
    # also you can use pydantic validation
    user_data: bytes = Body(min_length=1),
) -> ...:

StreamReader

Подробнее об объекте StreamReader можно узнать здесь.

from rapidy import StreamReader
from rapidy.http import post, Body

@post('/')
async def handler(
    user_data: StreamReader = Body(),
) -> ...:

---

Извлечение без валидации

Эти способы не рекомендуются.

Если отключить валидацию, параметр вернет базовую структуру aiohttp:

• Body(content_type=ContentType.stream) → bytes

Валидация pydantic для StreamReader не работает.

Способы отключения валидации

Отключение валидации с validate=False

Установите Body(validate=False).

@post('/')
async def handler(
    data: SomeBytesType = Body(validate=False, content_type=ContentType.stream),
) -> ...:

Отключение через Any

@post('/')
async def handler(
    data: Any = Body(content_type=ContentType.stream),
) -> ...:

Отключение типизации

Если не указывать тип, по умолчанию будет использован Any.

@post('/')
async def handler(
    data=Body(content_type=ContentType.stream),
) -> ...:

---

Значения по умолчанию

Если HTTP-запрос не содержит тела, параметр получит указанное значение по умолчанию (если оно задано).

Примеры использования

Указано значение по умолчанию

@post('/')
async def handler(
    data: bytes = Body(b'some_bytes', content_type=ContentType.stream),
    # or
    data: bytes = Body(default_factory=lambda: b'some_bytes', content_type=ContentType.stream),
) -> ...:

Опциональное тело запроса

@post('/')
async def handler(
    data: bytes | None = Body(content_type=ContentType.stream),
    # or
    data: Optional[bytes] = Body(content_type=ContentType.stream),
    # or
    data: Union[bytes, None] = Body(content_type=ContentType.stream),
) -> ...:

Значение по умолчанию для StreamReader задать нельзя.

Если попытаться установить default или default_factory для StreamReader, будет вызвана ошибка ParameterCannotUseDefaultError.

from rapidy import StreamReader
from rapidy.http import post, Body

@post('/')
async def handler(
    user_data: StreamReader = Body(default='SomeDefault'),
) -> ...:

------------------------------
Handler attribute with Type `Body` cannot have a default value.

Handler path: `<full_path>/main.py`
Handler name: `handler`
Attribute name: `data`
------------------------------

---

Как извлекаются сырые данные

Rapidy использует встроенные механизмы aiohttp.

Подробнее о aiohttp.Request и методах извлечения данных можно прочитать здесь.

bytes

Rapidy вызывает метод read объекта Request, а затем передает полученные данные в pydantic для валидации.

Как работает извлечение в Rapidy

async def extract_body_bytes(request: Request) -> Optional[bytes]:
    if not request.body_exists:
        return None

    return await request.read()

StreamReader

Rapidy использует атрибут content объекта Request и передает его напрямую в обработчик запроса, минуя валидацию pydantic.

Как работает извлечение в Rapidy

async def extract_body_stream(request: Request) -> Optional[StreamReader]:
    if not request.body_exists:
        return None

    return request.content