Source code for edri.api.headerkit.core
from __future__ import annotations
from collections import defaultdict
from collections.abc import Iterable, Iterator, Mapping
from typing import Any, TypeVar, overload
from .errors import DuplicateHeader, HeaderTooLarge, InvalidHeader
from .grammar import decode_latin1, validate_name_bytes, validate_value_bytes
from .policy import HeaderPolicy, ParseMode, UnknownHeaderMode, STRICT_POLICY
from .registry import FieldSpec, Registry
from .types import (
Accept,
Authorization,
CacheControl,
CombineMode,
Cookies,
DateHeader,
EntityTag,
EntityTagSet,
MediaType,
Range,
)
__all__ = ["Headers", "RawHeader"]
RawHeader = tuple[bytes, bytes]
T = TypeVar("T")
_MISSING = object()
[docs]
class Headers:
"""Lossless, lazy, typed HTTP header accessor for ASGI scopes.
The object keeps the original ASGI byte pairs intact while indexing by
lowercase header name. Semantic parsing is opt-in and lazy through the
registry.
"""
def __init__(
self,
raw: Iterable[RawHeader],
*,
registry: Registry | None = None,
policy: HeaderPolicy = STRICT_POLICY,
http_version: str = "1.1",
) -> None:
from .defaults import DEFAULT_REGISTRY
self.registry = registry or DEFAULT_REGISTRY
self.policy = policy
self.http_version = http_version
raw_original: list[RawHeader] = []
index: dict[bytes, list[bytes]] = defaultdict(list)
total = 0
lines = 0
for name, value in raw:
if not isinstance(name, bytes) or not isinstance(value, bytes):
raise InvalidHeader("ASGI headers must be (bytes, bytes) pairs")
lname = validate_name_bytes(name)
validate_value_bytes(value, allow_obs_text=policy.allow_obs_text)
lines += 1
if lines > policy.max_field_lines:
raise HeaderTooLarge("too many header field lines")
total += len(name) + len(value)
if total > policy.max_total_bytes:
raise HeaderTooLarge("headers exceed configured size limit")
raw_original.append((name, value))
index[lname].append(value)
self._raw = tuple(raw_original)
self._index: dict[bytes, tuple[bytes, ...]] = {
key: tuple(values) for key, values in index.items()
}
self._cache: dict[bytes, Any] = {}
[docs]
@classmethod
def from_scope(
cls,
scope: Mapping[str, Any],
*,
registry: Registry | None = None,
policy: HeaderPolicy = STRICT_POLICY,
) -> "Headers":
return cls(
scope.get("headers", ()),
registry=registry,
policy=policy,
http_version=scope.get("http_version", "1.1"),
)
def __contains__(self, name: bytes | str) -> bool:
return self.normalize_name(name) in self._index
def __iter__(self) -> Iterator[RawHeader]:
return iter(self._raw)
[docs]
@staticmethod
def normalize_name(name: bytes | str) -> bytes:
if isinstance(name, str):
try:
name = name.encode("ascii")
except UnicodeEncodeError as exc:
raise InvalidHeader(f"invalid non-ASCII header name: {name!r}") from exc
return validate_name_bytes(name)
[docs]
def decode_value(self, value: bytes) -> str:
text = decode_latin1(value)
return text.strip(" \t") if self.policy.strip_ows else text
@property
def raw(self) -> tuple[RawHeader, ...]:
"""Original ASGI byte pairs, with original casing/order/duplicates."""
return self._raw
[docs]
def raw_all(self, name: bytes | str) -> tuple[bytes, ...]:
return self._index.get(self.normalize_name(name), ())
[docs]
def all(self, name: bytes | str) -> tuple[str, ...]:
return tuple(self.decode_value(value) for value in self.raw_all(name))
@overload
def first(self, name: bytes | str) -> str | None: ...
@overload
def first(self, name: bytes | str, default: T) -> str | T: ...
[docs]
def first(self, name: bytes | str, default: Any = None) -> Any:
values = self.raw_all(name)
return self.decode_value(values[0]) if values else default
[docs]
def singleton(self, name: bytes | str) -> str | None:
raw = self.singleton_raw(name)
return None if raw is None else self.decode_value(raw)
[docs]
def singleton_raw(self, name: bytes | str) -> bytes | None:
lname = self.normalize_name(name)
values = self.raw_all(lname)
if not values:
return None
spec = self.registry.get(lname)
allow_identical = bool(spec and spec.allow_duplicate_identical)
if len(values) == 1:
return self._strip_ows_b(values[0])
normalized = tuple(self._strip_ows_b(v) for v in values)
if allow_identical and len(set(normalized)) == 1:
return normalized[0]
raise DuplicateHeader(f"duplicate singleton header: {lname.decode('ascii')}")
[docs]
def combined(self, name: bytes | str) -> bytes | None:
lname = self.normalize_name(name)
values = self.raw_all(lname)
if not values:
return None
spec = self.registry.get(lname)
combine = spec.combine if spec else CombineMode.OPAQUE
values = tuple(self._strip_ows_b(v) for v in values)
if combine is CombineMode.NEVER_COMBINE:
raise InvalidHeader(f"{lname.decode('ascii')} must not be comma-combined")
if combine is CombineMode.COOKIE:
return b"; ".join(v for v in values if v)
if combine is CombineMode.COMMA_LIST:
return b", ".join(v for v in values)
if combine is CombineMode.SINGLETON:
return self.singleton_raw(lname)
# Unknown or explicitly opaque field: no semantic combination by default.
if self.policy.unknown is UnknownHeaderMode.REJECT_DUPES and len(values) > 1:
raise DuplicateHeader(f"duplicate unknown header: {lname.decode('ascii')}")
if self.policy.unknown is UnknownHeaderMode.COMBINE:
return b", ".join(values)
if len(values) == 1:
return values[0]
raise DuplicateHeader(
f"cannot safely combine unknown/opaque header: {lname.decode('ascii')}"
)
[docs]
def parse(self, name: bytes | str) -> Any:
lname = self.normalize_name(name)
if lname in self._cache:
return self._cache[lname]
spec = self.registry.get(lname)
if spec is None or spec.parser is None:
value = self.combined(lname)
parsed = None if value is None else self.decode_value(value)
else:
parsed = spec.parser(self, lname)
self._cache[lname] = parsed
return parsed
[docs]
def list_members(self, name: bytes | str) -> tuple[str, ...]:
from .grammar import split_commas
value = self.combined(name)
if value is None:
return ()
return tuple(split_commas(self.decode_value(value)))
def _strip_ows_b(self, value: bytes) -> bytes:
return value.strip(b" \t") if self.policy.strip_ows else value
# Typed convenience properties.
@property
def accept(self) -> Accept:
return self.parse("accept")
@property
def accept_language(self) -> tuple[Any, ...]:
return self.parse("accept-language")
@property
def accept_encoding(self) -> tuple[Any, ...]:
return self.parse("accept-encoding")
@property
def cookies(self) -> Cookies:
return self.parse("cookie")
@property
def content_type(self) -> MediaType | None:
return self.parse("content-type")
@property
def content_length(self) -> int | None:
return self.parse("content-length")
@property
def cache_control(self) -> CacheControl:
return self.parse("cache-control")
@property
def authorization(self) -> Authorization | None:
return self.parse("authorization")
@property
def host(self) -> str | None:
return self.singleton("host")
@property
def etag(self) -> EntityTag | None:
return self.parse("etag")
@property
def if_none_match(self) -> EntityTagSet:
return self.parse("if-none-match")
@property
def range(self) -> Range | None:
return self.parse("range")
@property
def date(self) -> DateHeader | None:
return self.parse("date")