数据提取器

class litestar.data_extractors.ConnectionDataExtractor

基类：object

用于从 ASGIConnection、 Request 或 WebSocket 实例中提取数据的工具类。

__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

初始化 ConnectionDataExtractor Args: extract_body: 是否提取请求体（仅适用于请求）。 extract_client: 是否提取客户端（主机、端口）映射。 extract_content_type: 是否提取内容类型及任何选项。 extract_cookies: 是否提取 Cookie。 extract_headers: 是否提取请求头。 extract_method: 是否提取 HTTP 方法（仅适用于请求）。 extract_path: 是否提取路径。 extract_path_params: 是否提取路径参数。 extract_query: 是否提取查询参数。 extract_scheme: 是否提取 HTTP 协议方案。 obfuscate_headers: 需要混淆的请求头键。混淆后的值将被替换为 '*'。 obfuscate_cookies: 需要混淆的 Cookie 键。混淆后的值将被替换为 '*'。 parse_body: 是否解析请求体值或返回原始字节字符串（仅适用于请求）。 parse_query: 是否解析查询参数或返回原始字节字符串。 skip_parse_malformed_body: 如果请求体格式错误，是否跳过解析。

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

从连接中提取数据，返回一个包含值的字典。 Notes: - body 的值（如果存在）是一个未解析的协程，因此接收方应对其进行 await。 Args: connection: 一个 ASGI 连接或其子类。 Returns: 一个以字符串为键的字典，包含提取的值。

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

从 ASGIConnection 中提取协议方案 Args: connection: 一个 ASGIConnection 实例。 Returns: 连接的 scope["scheme"] 值

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

从 ASGIConnection 中提取客户端 Args: connection: 一个 ASGIConnection 实例。 Returns: 连接的 scope["client"] 值或默认值。

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

从 ASGIConnection 中提取路径 Args: connection: 一个 ASGIConnection 实例。 Returns: 连接的 scope["path"] 值

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

从 ASGIConnection 中提取标头 Args: connection: 一个 ASGIConnection 实例。 Returns: 包含连接标头的字典。

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

从 ASGIConnection 中提取 cookies Args: connection: 一个 ASGIConnection 实例。 Returns: 包含连接 cookies 的字典。

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

从 ASGIConnection 中提取查询参数 Args: connection: 一个 ASGIConnection 实例。 Returns: 包含连接已解析查询字符串的字典，或原始查询字节字符串。

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

从 ASGIConnection 中提取路径参数 Args: connection: 一个 ASGIConnection 实例。 Returns: 包含连接路径参数的字典。

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

从 ASGIConnection 中提取方法 Args: request: 一个 Request 实例。 Returns: 请求的 scope["method"] 值。

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

从 ASGIConnection 中提取 content-type Args: request: 一个 Request 实例。 Returns: 一个包含请求解析后的 'Content-Type' 头的元组。

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

从 ASGIConnection 中提取主体 Args: request: 一个 Request 实例。 Returns: 解析后的请求主体或原始字节字符串。

class litestar.data_extractors.ExtractedRequestData

基类：TypedDict

表示提取的请求数据的字典。

class litestar.data_extractors.ExtractedResponseData

基类：TypedDict

表示提取的响应数据的字典。

class litestar.data_extractors.ResponseDataExtractor

基类：object

从 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

使用选项初始化 ResponseDataExtractor。 Args: extract_body: 是否提取响应体。 extract_cookies: 是否提取 Cookie。 extract_headers: 是否提取请求头。 extract_status_code: 是否提取状态码。 obfuscate_cookies: 需要混淆的 Cookie 键。混淆后的值将被替换为 '*'。 obfuscate_headers: 需要混淆的请求头键。混淆后的值将被替换为 '*'。

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

从响应中提取数据，返回一个包含值的字典。 Args: messages: 一个元组，包含 HTTPResponseStartEvent 和 HTTPResponseBodyEvent。 Returns: 一个以字符串为键的字典，包含提取的值。

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

从 Message 中提取响应体 Args: messages: 一个元组，包含 HTTPResponseStartEvent 和 HTTPResponseBodyEvent。 Returns: 响应体，以字节字符串形式返回。

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

从 Message 中提取状态码 Args: messages: 一个包含 HTTPResponseStartEvent 和 HTTPResponseBodyEvent 的元组。 Returns: 响应的状态码。

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

从 Message 中提取标头 Args: messages: 一个元组，包含 HTTPResponseStartEvent 和 HTTPResponseBodyEvent。 Returns: 响应的标头字典。

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

从 Message 中提取 cookies Args: messages: 一个包含 HTTPResponseStartEvent 和 HTTPResponseBodyEvent 的元组。 Returns: 响应的 cookies 字典。