data_extractors

class litestar.data_extractors.ConnectionDataExtractor[源代码]

基类: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

参数:
  • 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.

备注

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

参数:

connection -- An ASGI connection or its subclasses.

返回:

A string keyed dictionary of extracted values.

static extract_scheme(connection: ASGIConnection[Any, Any, Any, Any]) str[源代码]

Extract the scheme from an ASGIConnection

参数:

connection -- An ASGIConnection instance.

返回:

The connection's scope["scheme"] value

static extract_client(connection: ASGIConnection[Any, Any, Any, Any]) tuple[str, int][源代码]

Extract the client from an ASGIConnection

参数:

connection -- An ASGIConnection instance.

返回:

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

参数:

connection -- An ASGIConnection instance.

返回:

The connection's scope["path"] value

extract_headers(connection: ASGIConnection[Any, Any, Any, Any]) dict[str, str][源代码]

Extract headers from an ASGIConnection

参数:

connection -- An ASGIConnection instance.

返回:

A dictionary with the connection's headers.

extract_cookies(connection: ASGIConnection[Any, Any, Any, Any]) dict[str, str][源代码]

Extract cookies from an ASGIConnection

参数:

connection -- An ASGIConnection instance.

返回:

A dictionary with the connection's cookies.

extract_query(connection: ASGIConnection[Any, Any, Any, Any]) Any[源代码]

Extract query from an ASGIConnection

参数:

connection -- An ASGIConnection instance.

返回:

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

参数:

connection -- An ASGIConnection instance.

返回:

A dictionary with the connection's path parameters.

static extract_method(request: Request[Any, Any, Any]) Method[源代码]

Extract the method from an ASGIConnection

参数:

request -- A Request instance.

返回:

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

参数:

request -- A Request instance.

返回:

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

参数:

request -- A Request instance.

返回:

Either the parsed request body or the raw byte-string.

class litestar.data_extractors.ExtractedRequestData[源代码]

基类:TypedDict

Dictionary representing extracted request data.

class litestar.data_extractors.ExtractedResponseData[源代码]

基类:TypedDict

Dictionary representing extracted response data.

class litestar.data_extractors.ResponseDataExtractor[源代码]

基类: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.

参数:
  • 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.

参数:

messages -- A tuple containing HTTPResponseStartEvent and HTTPResponseBodyEvent.

返回:

A string keyed dictionary of extracted values.

static extract_response_body(messages: tuple[HTTPResponseStartEvent, HTTPResponseBodyEvent]) bytes[源代码]

Extract the response body from a Message

参数:

messages -- A tuple containing HTTPResponseStartEvent and HTTPResponseBodyEvent.

返回:

The Response's body as a byte-string.

static extract_status_code(messages: tuple[HTTPResponseStartEvent, HTTPResponseBodyEvent]) int[源代码]

Extract a status code from a Message

参数:

messages -- A tuple containing HTTPResponseStartEvent and HTTPResponseBodyEvent.

返回:

The Response's status-code.

extract_headers(messages: tuple[HTTPResponseStartEvent, HTTPResponseBodyEvent]) dict[str, str][源代码]

Extract headers from a Message

参数:

messages -- A tuple containing HTTPResponseStartEvent and HTTPResponseBodyEvent.

返回:

The Response's headers dict.

extract_cookies(messages: tuple[HTTPResponseStartEvent, HTTPResponseBodyEvent]) dict[str, str][源代码]

Extract cookies from a Message

参数:

messages -- A tuple containing HTTPResponseStartEvent and HTTPResponseBodyEvent.

返回:

The Response's cookies dict.