data_extractors#

class litestar.data_extractors.ConnectionDataExtractor#

Bases: object

Utility class to extract data from an ASGIConnection, Request or WebSocket instance.

__init__(extract_body: bool = True, extract_client: bool = True, extract_content_type: bool = True, extract_cookies: bool = True, extract_headers: bool = True, extract_method: bool = True, extract_path: bool = True, extract_path_params: bool = True, extract_query: bool = True, extract_scheme: bool = True, obfuscate_cookies: set[str] | None = None, obfuscate_headers: set[str] | None = None, parse_body: bool = False, parse_query: bool = False, skip_parse_malformed_body: bool = False) → None#

Initialize ConnectionDataExtractor

Parameters:

extract_body¶ – Whether to extract body, (for requests only).
extract_client¶ – Whether to extract the client (host, port) mapping.
extract_content_type¶ – Whether to extract the content type and any options.
extract_cookies¶ – Whether to extract cookies.
extract_headers¶ – Whether to extract headers.
extract_method¶ – Whether to extract the HTTP method, (for requests only).
extract_path¶ – Whether to extract the path.
extract_path_params¶ – Whether to extract path parameters.
extract_query¶ – Whether to extract query parameters.
extract_scheme¶ – Whether to extract the http scheme.
obfuscate_headers¶ – headers keys to obfuscate. Obfuscated values are replaced with ‘*’.
obfuscate_cookies¶ – cookie keys to obfuscate. Obfuscated values are replaced with ‘*’.
parse_body¶ – Whether to parse the body value or return the raw byte string, (for requests only).
parse_query¶ – Whether to parse query parameters or return the raw byte string.
skip_parse_malformed_body¶ – Whether to skip parsing the body if it is malformed

__call__(connection: ASGIConnection[Any, Any, Any, Any]) → ExtractedRequestData#

Extract data from the connection, returning a dictionary of values.

Notes

The value for body - if present - is an unresolved Coroutine and as such should be awaited by the receiver.

Parameters:: connection¶ – An ASGI connection or its subclasses.
Returns:: A string keyed dictionary of extracted values.

static extract_scheme(connection: ASGIConnection[Any, Any, Any, Any]) → str#

Extract the scheme from an ASGIConnection

Parameters:: connection¶ – An ASGIConnection instance.
Returns:: The connection’s scope[“scheme”] value

static extract_client(connection: ASGIConnection[Any, Any, Any, Any]) → tuple[str, int]#

Extract the client from an ASGIConnection

Parameters:: connection¶ – An ASGIConnection instance.
Returns:: The connection’s scope[“client”] value or a default value.

static extract_path(connection: ASGIConnection[Any, Any, Any, Any]) → str#

Extract the path from an ASGIConnection

Parameters:: connection¶ – An ASGIConnection instance.
Returns:: The connection’s scope[“path”] value

extract_headers(connection: ASGIConnection[Any, Any, Any, Any]) → dict[str, str]#

Extract headers from an ASGIConnection

Parameters:: connection¶ – An ASGIConnection instance.
Returns:: A dictionary with the connection’s headers.

extract_cookies(connection: ASGIConnection[Any, Any, Any, Any]) → dict[str, str]#

Extract cookies from an ASGIConnection

Parameters:: connection¶ – An ASGIConnection instance.
Returns:: A dictionary with the connection’s cookies.

extract_query(connection: ASGIConnection[Any, Any, Any, Any]) → Any#

Extract query from an ASGIConnection

Parameters:: connection¶ – An ASGIConnection instance.
Returns:: Either a dictionary with the connection’s parsed query string or the raw query byte-string.

static extract_path_params(connection: ASGIConnection[Any, Any, Any, Any]) → dict[str, Any]#

Extract the path parameters from an ASGIConnection

Parameters:: connection¶ – An ASGIConnection instance.
Returns:: A dictionary with the connection’s path parameters.

static extract_method(request: Request[Any, Any, Any]) → Method#

Extract the method from an ASGIConnection

Parameters:: request¶ – A Request instance.
Returns:: The request’s scope[“method”] value.

static extract_content_type(request: Request[Any, Any, Any]) → tuple[str, dict[str, str]]#

Extract the content-type from an ASGIConnection

Parameters:: request¶ – A Request instance.
Returns:: A tuple containing the request’s parsed ‘Content-Type’ header.

async extract_body(request: Request[Any, Any, Any]) → Any#

Extract the body from an ASGIConnection

Parameters:: request¶ – A Request instance.
Returns:: Either the parsed request body or the raw byte-string.

class litestar.data_extractors.ExtractedRequestData#

Bases: TypedDict

Dictionary representing extracted request data.

class litestar.data_extractors.ExtractedResponseData#

Bases: TypedDict

Dictionary representing extracted response data.

class litestar.data_extractors.ResponseDataExtractor#

Bases: object

Utility class to extract data from a Message

__init__(extract_body: bool = True, extract_cookies: bool = True, extract_headers: bool = True, extract_status_code: bool = True, obfuscate_cookies: set[str] | None = None, obfuscate_headers: set[str] | None = None) → None#

Initialize ResponseDataExtractor with options.

Parameters:

extract_body¶ – Whether to extract the body.
extract_cookies¶ – Whether to extract the cookies.
extract_headers¶ – Whether to extract the headers.
extract_status_code¶ – Whether to extract the status code.
obfuscate_cookies¶ – cookie keys to obfuscate. Obfuscated values are replaced with ‘*’.
obfuscate_headers¶ – headers keys to obfuscate. Obfuscated values are replaced with ‘*’.

__call__(messages: tuple[HTTPResponseStartEvent, HTTPResponseBodyEvent]) → ExtractedResponseData#

Extract data from the response, returning a dictionary of values.

Parameters:: messages¶ – A tuple containing HTTPResponseStartEvent and HTTPResponseBodyEvent.
Returns:: A string keyed dictionary of extracted values.

static extract_response_body(messages: tuple[HTTPResponseStartEvent, HTTPResponseBodyEvent]) → bytes#

Extract the response body from a Message

Parameters:: messages¶ – A tuple containing HTTPResponseStartEvent and HTTPResponseBodyEvent.
Returns:: The Response’s body as a byte-string.

static extract_status_code(messages: tuple[HTTPResponseStartEvent, HTTPResponseBodyEvent]) → int#

Extract a status code from a Message

Parameters:: messages¶ – A tuple containing HTTPResponseStartEvent and HTTPResponseBodyEvent.
Returns:: The Response’s status-code.

extract_headers(messages: tuple[HTTPResponseStartEvent, HTTPResponseBodyEvent]) → dict[str, str]#

Extract headers from a Message

Parameters:: messages¶ – A tuple containing HTTPResponseStartEvent and HTTPResponseBodyEvent.
Returns:: The Response’s headers dict.

extract_cookies(messages: tuple[HTTPResponseStartEvent, HTTPResponseBodyEvent]) → dict[str, str]#

Extract cookies from a Message

Parameters:: messages¶ – A tuple containing HTTPResponseStartEvent and HTTPResponseBodyEvent.
Returns:: The Response’s cookies dict.