Skip to content

JSON

Reading the request body as JSON.

Description

JSON (JavaScript Object Notation) (MIME-type: application/json) is a text-based format for exchanging structured data, based on JavaScript. Today, JSON is language-independent and is used in various programming languages.

This section will show how to extract JSON from the request body and validate it using Rapidy.

Data Types in JSON

Object

An unordered set of key-value pairs. 

{
    "username": "User",
    "password": "myAwesomePass"
}

from pydantic import BaseModel
from rapidy.http import post, Body, ContentType

class UserData(BaseModel):
    username: str
    password: str

@post('/')
async def handler(
    user_data: UserData = Body(),
    # or
    user_data: UserData = Body(content_type=ContentType.json),
    # or
    user_data: UserData = Body(content_type='application/json'),
) -> ...:

Sending with curl 
curl -X POST \
-H "Content-Type: application/json" \
-d '{"username": "User", "password": "myAwesomePass"}' \
http://127.0.0.1:8080

Array

[
    {"username": "User1", "password": "myAwesomePass1"},
    {"username": "User2", "password": "myAwesomePass2"}
]

from pydantic import BaseModel
from rapidy.http import post, Body

class UserData(BaseModel):
    username: str
    password: str

@post('/')
async def handler(
    users: list[UserData] = Body(),
) -> ...:

Sending with curl 
curl -X POST \
-H "Content-Type: application/json" \
-d '[{"username": "user1", "password": "password1"}, {"username": "user2", "password": "password2"}]' \
http://127.0.0.1:8080

Number

An integer or floating-point value. 

111

from rapidy.http import post, Body

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

Sending with curl 
curl -X POST \
-H "Content-Type: application/json" \
-d '"111"' \
http://127.0.0.1:8080

When sending a string in JSON format, additional escaping is required: "111"

Literals

true (boolean true), false (boolean false), and null (absence of a value). 

true
false
null

from rapidy.http import post, Body

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

Sending with curl 

curl -X POST \
-H "Content-Type: application/json" \
-d 'true' \
http://127.0.0.1:8080

curl -X POST \
-H "Content-Type: application/json" \
-d 'false' \
http://127.0.0.1:8080

curl -X POST \
-H "Content-Type: application/json" \
-d 'null' \
http://127.0.0.1:8080

String

"SomeString"

from rapidy.http import post, Body

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

Sending with curl 
curl -X POST \
-H "Content-Type: application/json" \
-d '"SomeString"' \
http://127.0.0.1:8080

When sending a string in JSON format, escaping is required: \"SomeString\"

Custom JSON Decoder

By default, Rapidy uses json.loads without parameters to decode incoming JSON.

Equivalent examples:

from rapidy.http import post, Body

@post('/')
async def handler(
    data: str = Body(),
) -> ...:
or 
import json
from rapidy.http import post, Body

@post('/')
async def handler(
    data: str = Body(json_decoder=json.loads),
) -> ...:

To use a custom decoder, pass a callable object that takes a str as the json_decoder parameter.

Expected type: Callable[[str], Any]

Example with a custom decoder:

from typing import Any
from rapidy.http import post, Body

def custom_json_decoder(data: str) -> ...:
    ...

@post('/')
async def handler(
    data: Any = Body(json_decoder=custom_json_decoder),
) -> ...:

If you need to use json.loads with parameters, use functools.partial: 

import json
from functools import partial
from typing import Any, OrderedDict
from rapidy.http import post, Body

decoder = partial(json.loads, object_pairs_hook=OrderedDict)

@post('/')
async def handler(
    data: Any = Body(json_decoder=decoder),
) -> ...:

Extraction Without Validation

Disabling validation is not recommended.

If validation is disabled, the parameter will contain data in the form it was unpacked by the JSON decoder.

Ways to Disable Validation

Explicit Disabling

from pydantic import BaseModel
from rapidy.http import post, Body

class BodyData(BaseModel):
    ...

@post('/')
async def handler(
    data: BodyData = Body(validate=False),
) -> ...:

Using Any

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

No Type Annotation

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

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

Default Values

If an HTTP request does not contain a body, the parameter will receive the specified default value (if set).

Usage Examples

Default Value Specified

from pydantic import BaseModel
from rapidy.http import post, Body

class BodyData(BaseModel):
    ...

@post('/')
async def handler(
    data: BodyData = Body('some_data'),
    # or
    data: BodyData = Body(default_factory=lambda: 'some_data'),
) -> ...:

Optional Request Body

from pydantic import BaseModel
from rapidy.http import post, Body

class BodyData(BaseModel):
    ...

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

Extracting Raw Data

Rapidy uses the json method of the Request object to obtain data and passes it to Pydantic for validation.

If the data cannot be extracted as JSON, an ExtractError will be returned: 

{
    "errors": [
        {
            "type": "ExtractError",
            "loc": [
                "body"
            ],
            "msg": "Failed to extract body data as Json: <error_description>"
        }
    ]
}

How data extraction works in Rapidy

async def extract_body_json(request: Request, body_field_info: Body) -> Optional[DictStrAny]:
    if not request.body_exists:
        return None

    return await request.json(loads=body_field_info.json_decoder)

Rapidy uses built-in aiohttp mechanisms for data extraction.

More details about the aiohttp.Request object and methods for extracting data from it can be found here.

If a parameter is annotated as bytes or StreamReader, data is extracted differently.

More details about the StreamReader object can be found here.

bytes

from rapidy.http import post, Body

@post('/')
async def handler(
    user_data: bytes = Body(),
    # also you can use pydantic validation
    user_data: bytes = Body(min_length=1),
) -> ...:
Rapidy Internal Code 
async def extract_body_bytes(request: Request) -> Optional[bytes]:
    if not request.body_exists:
        return None

    return await request.read()

StreamReader

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

@post('/')
async def handler(
    user_data: StreamReader = Body(),
) -> ...:
Rapidy Internal Code 
async def extract_body_stream(request: Request) -> Optional[StreamReader]:
    if not request.body_exists:
        return None

    return request.content

Validation with Pydantic is not supported for StreamReader.

A default value cannot be set for StreamReader.

If you attempt to set a default value for Body with a StreamReader annotation using default or default_factory, a ParameterCannotUseDefaultError will be raised. 

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`
------------------------------