Binary

Reading the request body as a byte sequence.

Description

Binary (MIME-type: application/octet-stream) — a binary data type.

Rapidy allows extracting any data with content_type as a sequence of bytes.

Simply annotate it as bytes or StreamReader.

Why is this useful?

This is useful when you need to explicitly restrict the type of received data and then process it in binary format.

There are only two data types that can be extracted regardless of content_type: bytes and 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

You can learn more about the StreamReader object here.

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

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

---

Extraction without validation

Disabling validation is not recommended.

If validation is disabled, the parameter will contain the base aiohttp structure:

• Body(content_type=ContentType.text_plain) → bytes

pydantic validation does not work for StreamReader.

Ways to disable validation

Explicit Disabling

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

Using Any

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

No Type Annotation

If no type is specified, Any will be set by default.

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

---

Default values

If the HTTP request body is not provided, a default value (if set) will be used.

Usage Examples

Default Value Specified

@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),
) -> ...:

Optional Request Body

@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),
) -> ...:

A default value cannot be set for StreamReader.

Attempting to set a default value for StreamReader using default or default_factory will raise a 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`
------------------------------

---

How raw data is extracted

Rapidy uses built-in aiohttp data extraction mechanisms.

You can learn more about the aiohttp.Request object and its data extraction methods here.

bytes

Rapidy calls the read method on the Request object and then passes the retrieved data to pydantic for validation.

How data extraction works in Rapidy

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

    return await request.read()

StreamReader

Rapidy accesses the content attribute of the Request object and passes it directly to the request handler, bypassing pydantic validation.

How data extraction works in Rapidy

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

    return request.content