from __future__ import annotations
from re import compile
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import Iterable, Iterator, Mapping
__all__ = [
"Accept", "AcceptItem", "Authorization", "ByteRange", "CacheControl",
"CombineMode", "Cookies", "DateHeader", "EntityTag", "EntityTagSet",
"Forwarded", "ForwardedElement", "MediaType", "Range", "Weighted",
]
# Chunked-cookie suffix: name.0, name.1, …
_COOKIE_CHUNK_RE = compile(r"^(?P<base>.+)\.(?P<index>\d+)$")
[docs]
class CombineMode(str, Enum):
OPAQUE = "opaque"
SINGLETON = "singleton"
COMMA_LIST = "comma-list"
NEVER_COMBINE = "never-combine"
COOKIE = "cookie"
[docs]
@dataclass(frozen=True, slots=True)
class Weighted:
value: str
q: float = 1.0
params: Mapping[str, str] | None = None
order: int = 0
@property
def acceptable(self) -> bool:
return self.q > 0
[docs]
@dataclass(frozen=True, slots=True)
class AcceptItem:
media: MediaType
q: float = 1.0
accept_ext: Mapping[str, str] | None = None
order: int = 0
@property
def value(self) -> str:
return self.media.essence
@property
def specificity(self) -> int:
# exact/exact > exact/* > */*
return int(self.media.type != "*") + int(self.media.subtype != "*")
@property
def acceptable(self) -> bool:
return self.q > 0
[docs]
@dataclass(frozen=True, slots=True)
class Accept:
items: tuple[AcceptItem, ...]
def __iter__(self) -> Iterator[AcceptItem]:
return iter(self.items)
def __bool__(self) -> bool:
return bool(self.items)
[docs]
def best_match(self, offers: Iterable[str], default: str | None = None) -> str | None:
best_offer: str | None = None
best_key: tuple[float, int, int, int] | None = None
for offer_order, offer in enumerate(offers):
offer_l = offer.lower()
for item in self.items:
if not item.acceptable or not item.media.matches(offer_l):
continue
key = (item.q, item.specificity, -item.order, -offer_order)
if best_key is None or key > best_key:
best_key = key
best_offer = offer
return best_offer if best_offer is not None else default
@dataclass(frozen=True, slots=True)
class MediaRange:
value: str
params: Mapping[str, str]
[docs]
@dataclass(frozen=True, slots=True)
class CacheControl:
directives: Mapping[str, str | bool]
def __contains__(self, key: str) -> bool:
return key.lower() in self.directives
[docs]
def get(self, key: str, default: str | bool | None = None) -> str | bool | None:
return self.directives.get(key.lower(), default)
@property
def no_cache(self) -> bool:
return "no-cache" in self.directives
@property
def no_store(self) -> bool:
return "no-store" in self.directives
@property
def max_age(self) -> int | None:
value = self.directives.get("max-age")
if isinstance(value, str) and value.isdecimal():
return int(value)
return None
[docs]
@dataclass(frozen=True, slots=True)
class Cookies:
pairs: tuple[tuple[str, str], ...]
[docs]
def get_all(self, name: str) -> tuple[str, ...]:
return tuple(value for key, value in self.pairs if key == name)
[docs]
def get(self, name: str, default: str | None = None) -> str | None:
values = self.get_all(name)
return values[-1] if values else default
[docs]
def as_dict(self) -> dict[str, str]:
return {key: value for key, value in self.pairs}
[docs]
def dechunked(self) -> "Cookies":
"""Reassemble Auth.js / NextAuth chunked cookies.
When a cookie value exceeds the ~4 KB browser limit, Auth.js splits it
into ``name.0``, ``name.1``, … and the browser sends each chunk back as
a separate cookie. This rejoins a contiguous family starting at ``.0``
into a single ``name`` cookie (values concatenated in index order).
A non-chunked cookie of the same base name takes precedence (its chunk
siblings are then left untouched). Chunk families with a missing ``.0``
or a gap keep their unconsumed members as literal cookies, so nothing is
silently dropped.
"""
present = {key for key, _ in self.pairs}
result: list[tuple[str, str]] = []
chunks: dict[str, dict[int, str]] = {}
for key, value in self.pairs:
match = _COOKIE_CHUNK_RE.match(key)
if match and match.group("base") not in present:
chunks.setdefault(match.group("base"), {})[int(match.group("index"))] = value
else:
result.append((key, value))
for base, parts in chunks.items():
if 0 not in parts:
result.extend((f"{base}.{i}", parts[i]) for i in sorted(parts))
continue
index = 0
buffer: list[str] = []
while index in parts:
buffer.append(parts.pop(index))
index += 1
result.append((base, "".join(buffer)))
result.extend((f"{base}.{i}", parts[i]) for i in sorted(parts))
return Cookies(tuple(result))
[docs]
@dataclass(frozen=True, slots=True)
class Authorization:
scheme: str
credentials: str
[docs]
@dataclass(frozen=True, slots=True)
class EntityTag:
tag: str
weak: bool = False
[docs]
@dataclass(frozen=True, slots=True)
class ByteRange:
start: int | None
end: int | None
[docs]
@dataclass(frozen=True, slots=True)
class Range:
unit: str
ranges: tuple[ByteRange, ...]
[docs]
@dataclass(frozen=True, slots=True)
class ForwardedElement:
params: Mapping[str, str]
[docs]
@dataclass(frozen=True, slots=True)
class Forwarded:
elements: tuple[ForwardedElement, ...]