Request Class

The Request class provides access to HTTP request data including headers, body, and path parameters.

Request

Request(
    scope: dict[str, Any],
    receive: Callable[[], Awaitable[dict[str, Any]]],
)

HTTP request object providing access to request data.

This class encapsulates the ASGI scope and receive callable to provide a convenient interface for accessing request data including headers, body content, and path parameters.

Parameters:

Name	Type	Description	Default
scope	dict[str, Any]	ASGI scope dictionary containing request metadata	required
receive	Callable[[], Awaitable[dict[str, Any]]]	ASGI receive callable for getting request body	required

Attributes:

Name	Type	Description
scope		The ASGI scope dictionary
receive		The ASGI receive callable
path_params	dict[str, str]	Dictionary of extracted path parameters
headers	dict[str, str]	Dictionary of request headers

Source code in src/artanis/request.py

def __init__(
    self, scope: dict[str, Any], receive: Callable[[], Awaitable[dict[str, Any]]]
) -> None:
    self.scope = scope
    self.receive = receive
    self._body: bytes | None = None
    self.path_params: dict[
        str, str
    ] = {}  # For middleware access to path parameters
    # Convert ASGI headers (list of byte tuples) to string dict
    raw_headers = scope.get("headers", [])
    self.headers: dict[str, str] = {
        name.decode().lower(): value.decode() for name, value in raw_headers
    }

Functions

body async

body() -> bytes

Get the request body as bytes.

Reads and caches the complete request body from the ASGI receive callable. The body is cached after the first call to avoid multiple reads.

Returns:

Type	Description
bytes	The complete request body as bytes

Source code in src/artanis/request.py

async def body(self) -> bytes:
    """Get the request body as bytes.

    Reads and caches the complete request body from the ASGI receive callable.
    The body is cached after the first call to avoid multiple reads.

    Returns:
        The complete request body as bytes
    """
    if self._body is None:
        body_parts = []
        while True:
            message = await self.receive()
            if message["type"] == "http.request":
                body_parts.append(message.get("body", b""))
                if not message.get("more_body", False):
                    break
        self._body = b"".join(body_parts)
    return self._body

json async

json() -> Any

Parse request body as JSON.

Reads the request body and parses it as JSON data.

Returns:

Type	Description
Any	Parsed JSON data (dict, list, or other JSON-serializable types)

Raises:

Type	Description
ValidationError	If the body is not valid JSON

Source code in src/artanis/request.py

async def json(self) -> Any:
    """Parse request body as JSON.

    Reads the request body and parses it as JSON data.

    Returns:
        Parsed JSON data (dict, list, or other JSON-serializable types)

    Raises:
        ValidationError: If the body is not valid JSON
    """
    try:
        body = await self.body()
        return json.loads(body.decode())
    except json.JSONDecodeError as e:
        raise ValidationError(
            message="Invalid JSON in request body",
            field="body",
            value=body.decode() if len(body) < 200 else body.decode()[:200] + "...",
            validation_errors={"json_error": str(e)},
        )