API reference¶

def last_page_offset(total: int, limit: int) -> int:
    """The zero-based offset of the last page for ``total`` items at page size ``limit``.

    Zero when there are no items. ``limit`` must be positive. Shared so callers
    deriving their own ``last`` cursor don't re-spell the rounding math.
    """
    return ((total - 1) // limit) * limit if total > 0 else 0

def paginate(
    *,
    next_token: str | None = None,
    prev_token: str | None = None,
    limit: int | None = None,
    first: bool = False,
    last_token: str | None = None,
    self_: bool = False,
    method: str = 'GET',
    body: Mapping[str, Any] | None = None,
    type: str | None = MediaType.JSON,
    headers: Mapping[str, str | list[str]] | None = None,
    token_param: str = 'token',  # noqa: S107
    limit_param: str = 'limit',
    **link_fields: Any,
) -> list[Link]:
    """Return token-based pagination links for the values provided (deferred).

    Args:
        next_token: Token for the next page; emits a ``next`` link when set.
        prev_token: Token for the previous page; emits a ``prev`` link when set.
        limit: Page size to carry on every emitted link. When ``None`` (the default),
            ``limit_param`` is left out of the merge entirely, so a GET link preserves
            whatever limit (if any) is already on the current URL, and a POST link
            leaves it out of (rather than popping it from) the caller's ``body``.
        first: Emit a ``first`` link to the un-tokened first page when ``True``.
        last_token: Token for the last page; emits a ``last`` link when set.
        self_: Emit a ``self`` link to the current request when ``True``. Note that for
            ``method='POST'`` the ``self`` link's body is exactly the caller's ``body``
            — it does *not* bake in the current page's token, so re-POSTing it returns
            page one unless the caller includes the current token in ``body``.
        method: ``'GET'`` (default) carries the token in the query string; ``'POST'``
            carries it in the request **body** (merged into ``body``) and keeps the
            current URL as the href — the form a stateless server needs.
        body: Base request body for ``method='POST'`` (e.g. the search criteria); the
            pagination params are merged into a copy of it per link.
        type: The ``type`` (target media type) set on every emitted link.
        headers: Optional ``headers`` member set on every emitted link.
        token_param: Query/body parameter name carrying the token.
        limit_param: Query/body parameter name carrying the limit.
        **link_fields: Any further ``Link`` members (e.g. ``title``) applied to every
            emitted link.

    The ``next``/``prev`` links lead the list (unchanged from earlier releases); any
    ``first``/``last``/``self`` links follow.
    """
    style = _LinkStyle(method=method, body=body, type=type, headers=headers, extra=link_fields)

    def link(rel: str, token: str | None) -> Link:
        params: dict[str, object] = {token_param: token}
        if limit is not None:
            params[limit_param] = limit
        return _page_link(rel, params, style)

    links: list[Link] = []
    if next_token is not None:
        links.append(link(Rel.NEXT, next_token))
    if prev_token is not None:
        links.append(link(Rel.PREV, prev_token))
    if first:
        # The first page is the collection with no token; an explicit limit still
        # overrides, but an absent one leaves the current URL's limit untouched.
        first_params: dict[str, object] = {token_param: None}
        if limit is not None:
            first_params[limit_param] = limit
        links.append(_page_link(Rel.FIRST, first_params, style))
    if last_token is not None:
        links.append(link(Rel.LAST, last_token))
    if self_:
        # self repeats the current request unchanged (no token rewrite).
        links.append(_page_link(Rel.SELF, {}, style))
    return links

def paginate_offset(
    *,
    offset: int,
    limit: int,
    total: int | None = None,
    self_: bool = True,
    method: str = 'GET',
    body: Mapping[str, Any] | None = None,
    type: str | None = MediaType.JSON,
    headers: Mapping[str, str | list[str]] | None = None,
    offset_param: str = 'offset',
    limit_param: str = 'limit',
    **link_fields: Any,
) -> list[Link]:
    """Return offset/limit pagination links, derived from the current page (deferred).

    Emits ``self``/``first``/``prev``/``next``/``last`` as the position warrants:
    ``first``/``prev`` only when ``offset > 0``; ``next`` whenever ``total`` is unknown
    or another page follows; ``last`` only when ``total`` is known and differs from the
    current page. Each link is canonical — it carries the explicit ``offset``/``limit``.

    Accepts the same ``method``/``body``/``type``/``headers``/``**link_fields``
    pass-through as :func:`paginate` (so offset paging can also ride a POST body).

    Args:
        offset: The current page's zero-based item offset.
        limit: The page size (must be positive).
        total: Total matching items, if known; enables the ``last`` link and lets
            ``next`` stop at the end.
        self_: Emit a ``self`` link to the current (canonical) page when ``True``.
        offset_param: Query/body parameter name carrying the offset.
        limit_param: Query/body parameter name carrying the limit.

    Raises:
        ValueError: If ``limit`` is not positive or ``offset`` is negative.
    """
    if limit <= 0:
        raise ValueError('limit must be positive')
    if offset < 0:
        raise ValueError('offset must not be negative')

    style = _LinkStyle(method=method, body=body, type=type, headers=headers, extra=link_fields)

    def at(rel: str, page_offset: int) -> Link:
        params = {offset_param: page_offset, limit_param: limit}
        return _page_link(rel, params, style)

    links: list[Link] = []
    if self_:
        links.append(at(Rel.SELF, offset))
    if offset > 0:
        links.append(at(Rel.FIRST, 0))
        links.append(at(Rel.PREV, max(0, offset - limit)))
    if total is None or offset + limit < total:
        links.append(at(Rel.NEXT, offset + limit))
    if total is not None:
        last_offset = last_page_offset(total, limit)
        if last_offset != offset:
            links.append(at(Rel.LAST, last_offset))
    return links

def encode_cursor(payload: Mapping[str, Any]) -> str:
    """Encode an arbitrary token ``payload`` as one opaque, URL-safe cursor string.

    The payload (any JSON-serializable mapping — e.g. ``{'after_id': 42}``) is
    serialized to compact JSON and base64url-encoded without padding, so the result
    is safe to drop straight into a ``next``/``prev`` token. Round-trips through
    :func:`decode_cursor`. The cursor is **opaque, not secret** — it is encoded, not
    signed or encrypted, so never trust it without validating the decoded contents.
    """
    raw = json.dumps(payload, separators=(',', ':'), sort_keys=True).encode('utf-8')
    return base64.urlsafe_b64encode(raw).rstrip(b'=').decode('ascii')

def decode_cursor(token: str, *, parameter: str = 'token') -> dict[str, Any]:
    """Decode a cursor produced by :func:`encode_cursor` back into its payload.

    A malformed or non-object cursor raises :class:`~gazebo.params.ParamError` (which
    the FastAPI glue renders as a ``400`` problem) carrying ``parameter`` — treat a
    bad client-supplied cursor as a client error, not a 500. ``parameter`` defaults to
    ``'token'`` to match :func:`paginate`'s default ``token_param``; pass the actual
    query/body parameter name if the caller uses a different one (e.g. ``'cursor'``).
    """
    padded = token + '=' * (-len(token) % 4)
    try:
        raw = base64.urlsafe_b64decode(padded.encode('ascii'))
        data = json.loads(raw)
    except (ValueError, UnicodeError) as exc:
        raise ParamError(parameter, 'malformed cursor') from exc
    if not isinstance(data, dict):
        raise ParamError(parameter, 'cursor must encode an object')
    return data

def format_link_header(
    links: Iterable[Mapping[str, Any]],
    *,
    rels: Sequence[str] | None = NAV_RELS,
    max_links: int = DEFAULT_MAX_LINKS,
) -> str:
    """Serialize resolved ``links`` into an RFC 8288 ``Link:`` header value.

    Args:
        links: Resolved link mappings — each a dict with at least ``href`` and
            ``rel`` (the shape gazebo serializes into a JSON body). A link missing
            either is skipped, and a callable (unresolved) href cannot appear here.
        rels: The relations to include, in any order; a link whose ``rel`` is not in
            this set is dropped. ``None`` disables filtering (include every rel) —
            use with care, since that can include per-item links.
        max_links: Stop after this many links (after filtering), guarding header size.

    Returns:
        The header value (links joined by ``, ``), or ``''`` when nothing qualifies —
        callers should then omit the header entirely rather than send an empty one.
    """
    allowed = None if rels is None else frozenset(rels)
    out: list[str] = []
    for link in links:
        if allowed is not None and link.get('rel') not in allowed:
            continue
        formatted = _format_one(link)
        if formatted is None:
            continue
        out.append(formatted)
        if len(out) >= max_links:
            break
    return ', '.join(out)

def etag_for(value: Any, *, weak: bool = True) -> str:
    """Derive an ``ETag`` from ``value`` (a model, mapping, str, or bytes).

    The value is reduced to canonical bytes — a pydantic model via
    ``model_dump_json(by_alias=True)``, anything else via sorted-key JSON — and hashed
    (SHA-256). The result is a quoted entity-tag, prefixed ``W/`` when ``weak`` (the
    default).

    Note:
        A model carrying deferred (callable-href) links only serializes inside an
        active request context; outside one, ETag such a model from its underlying
        data rather than the link-bearing envelope.
    """
    digest = hashlib.sha256(_canonical_bytes(value)).hexdigest()
    etag = f'"{digest}"'
    return f'W/{etag}' if weak else etag

def http_date(value: datetime) -> str:
    """Format ``value`` as an IMF-fixdate HTTP date (e.g. for ``Last-Modified``)."""
    if value.tzinfo is None:
        value = value.replace(tzinfo=UTC)
    return format_datetime(value, usegmt=True)

def parse_http_date(value: str) -> datetime | None:
    """Parse an HTTP date header into an aware ``datetime`` (``None`` if malformed)."""
    try:
        parsed = parsedate_to_datetime(value)
    except (TypeError, ValueError):
        return None
    return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC)

def if_none_match_satisfied(etag: str, header: str) -> bool:
    """Whether ``etag`` matches an ``If-None-Match`` header value (weak comparison).

    ``*`` matches any current entity. Per RFC 7232, ``If-None-Match`` uses the *weak*
    comparison function, so the ``W/`` prefix is ignored on both sides. The header's
    comma-separated entity-tags are parsed with a quoted-string-aware pattern (rather
    than a plain ``split(',')``) since RFC 7232's ``etagc`` grammar permits a literal
    comma inside a quoted entity-tag; a candidate that isn't a well-formed quoted
    entity-tag is simply not matched.
    """
    header = header.strip()
    if header == '*':
        return True
    current = _normalize_etag(etag)
    candidates = _ETAG_PATTERN.findall(header)
    return any(current == _normalize_etag(candidate) for candidate in candidates)

def is_not_modified(
    *,
    method: str = 'GET',
    etag: str | None = None,
    last_modified: datetime | None = None,
    if_none_match: str | None = None,
    if_modified_since: str | None = None,
) -> bool:
    """Evaluate the conditional-GET preconditions; ``True`` means respond ``304``.

    Only ``GET``/``HEAD`` are eligible. ``If-None-Match`` takes precedence over
    ``If-Modified-Since`` (which is ignored entirely when the former is present, per
    RFC 7232 §3.3). HTTP dates carry one-second resolution, so ``last_modified`` is
    truncated to whole seconds before comparison.
    """
    if method.upper() not in _CONDITIONAL_METHODS:
        return False
    if if_none_match is not None:
        return etag is not None and if_none_match_satisfied(etag, if_none_match)
    if if_modified_since is not None and last_modified is not None:
        since = parse_http_date(if_modified_since)
        if since is None:
            return False
        lm = (
            last_modified
            if last_modified.tzinfo is not None
            else last_modified.replace(tzinfo=UTC)
        )
        return lm.replace(microsecond=0) <= since
    return False

@contextmanager
def use_context(ctx: RequestContext) -> Iterator[RequestContext]:
    """Bind ``ctx`` as the active request context for the duration of the block.

    Uses an explicit ``reset`` in ``finally`` so it is correct on every supported
    Python version (``Token`` only became a context manager in 3.14).
    """
    token: Token[RequestContext | None] = link_context.set(ctx)
    try:
        yield ctx
    finally:
        link_context.reset(token)

def resolve_context(info_context: Any = None) -> RequestContext | None:
    """Find the active request context.

    Resolution order: the :data:`link_context` ContextVar first, then a
    ``request``/``context`` entry in a pydantic serialization ``info.context``
    mapping (the manual-dump / test escape hatch).
    """
    ctx = link_context.get(None)
    if ctx is not None:
        return ctx
    if isinstance(info_context, Mapping):
        candidate = info_context.get('request') or info_context.get('context')
        if candidate is not None:
            return candidate
    return None

def merge_params(params: dict[str, Any], overrides: Mapping[str, object]) -> None:
    """Merge ``overrides`` into ``params`` in place.

    A ``None`` value in ``overrides`` removes the key from ``params``; any other value
    sets it. Shared by :func:`with_query` (query-string overrides) and pagination's POST
    body merge, so the two identical "None removes, else sets" merges aren't re-spelled.
    """
    for key, value in overrides.items():
        if value is None:
            params.pop(key, None)
        else:
            params[key] = value