Source code for edri.api.headerkit.response

from __future__ import annotations

from collections.abc import Iterable
from datetime import datetime, timezone
from email.utils import format_datetime
from http.cookies import SimpleCookie

from .core import RawHeader
from .errors import InvalidHeader
from .grammar import quote, serialize_params, validate_name_bytes, validate_value_bytes

__all__ = ["ResponseHeaders"]


def _name(name: bytes | str) -> bytes:
    if isinstance(name, str):
        name = name.encode("ascii")
    return validate_name_bytes(name)


def _value(value: bytes | str | int) -> bytes:
    if isinstance(value, bytes):
        raw = value
    elif isinstance(value, int):
        raw = str(value).encode("ascii")
    else:
        raw = value.encode("latin-1")
    validate_value_bytes(raw)
    return raw


[docs] class ResponseHeaders: """Builder for ASGI response header byte pairs. It deliberately treats Set-Cookie as append-only and never coalesces it. """ def __init__(self, raw: Iterable[RawHeader] = ()) -> None: self._raw: list[RawHeader] = [] for name, value in raw: self.append(name, value)
[docs] def append(self, name: bytes | str, value: bytes | str | int) -> None: self._raw.append((_name(name), _value(value)))
[docs] def set(self, name: bytes | str, value: bytes | str | int) -> None: lname = _name(name) if lname == b"set-cookie": raise InvalidHeader("use set_cookie()/append() for Set-Cookie; it is not a singleton") self.remove(lname) self.append(lname, value)
[docs] def remove(self, name: bytes | str) -> None: lname = _name(name) self._raw = [(n, v) for n, v in self._raw if n != lname]
[docs] def extend(self, raw: Iterable[RawHeader]) -> None: for name, value in raw: self.append(name, value)
[docs] def set_content_type(self, media_type: str, **params: str) -> None: suffix = serialize_params(params) value = media_type if not suffix else f"{media_type}; {suffix}" self.set("content-type", value)
[docs] def set_date(self, name: str, value: datetime | None = None) -> None: if value is None: value = datetime.now(timezone.utc) if value.tzinfo is None: value = value.replace(tzinfo=timezone.utc) self.set(name, format_datetime(value.astimezone(timezone.utc), usegmt=True))
[docs] def asgi(self) -> list[RawHeader]: return list(self._raw)
def __iter__(self): return iter(self._raw) def __contains__(self, name: bytes | str) -> bool: lname = _name(name).lower() return any(n.lower() == lname for n, _ in self._raw)