Module

_headers

Shared header lookup utilities.

Single-pass header extraction for ASGI headers (list or tuple of (name, value) byte pairs). Consolidates the 7 copies of _get_header scattered across worker, sync_worker, async_pool, and handler modules.

Functions

is_websocket_upgrade 1 bool
Check if the request is a WebSocket upgrade. Detects ``Connection: Upgrade`` +…
def is_websocket_upgrade(request: RequestReceived) -> bool

Check if the request is a WebSocket upgrade.

DetectsConnection: Upgrade + Upgrade: websocketheaders. Works with both pre-lowered (fast parser) and raw-cased (h11) header names.

Parameters
Name Type Description
request RequestReceived
Returns
bool
get_header 2 bytes | None
Get a header value by lowercase name. Single linear scan — use when only one h…
def get_header(headers: Sequence[tuple[bytes, bytes]], name: bytes) -> bytes | None

Get a header value by lowercase name.

Single linear scan — use when only one header is needed.

Parameters
Name Type Description
headers Sequence[tuple[bytes, bytes]]

ASGI headers (list or tuple of (name, value) pairs).

name bytes

Header name to find (compared case-insensitively).

Returns
bytes | None