Module adcp.decisioning.pagination

Framework-managed cursor pagination for list responses.

When an adopter sets auto_paginate=True on :class:DecisioningCapabilities, the handler intercepts get_products, calls the adopter's implementation to retrieve the full product set, and slices the result to the requested page.

The cursor is an HMAC-SHA-256-signed JSON envelope::

base64url( JSON({ "p": JSON({"v":1,"o":<offset>,"qh":<filter-hash>}), "s":<hex-sig> }) )

The qh field is a SHA-256 fingerprint of the request parameters (excluding pagination itself), so if a buyer changes filters between pages the framework returns INVALID_REQUEST with field="pagination.cursor" rather than silently serving the wrong page slice.

Secret management. The HMAC key defaults to a per-process random secret (stable within the process, different across restarts). Stale cursors from a prior process return INVALID_REQUEST — this is the correct behaviour for stateless sellers. Horizontally-scaled or stateless sellers that need cursors to survive process restarts MUST set ADCP_PAGINATION_SECRET in the environment.

Functions

def apply_framework_pagination(response: Any, pagination: Any, query_hash: str, secret: bytes | None = None) ‑> Any

Expand source code

def apply_framework_pagination(
    response: Any,
    pagination: Any,
    query_hash: str,
    secret: bytes | None = None,
) -> Any:
    """Slice a full-list ``GetProductsResponse`` to the requested page.

    Called by the handler post-adapter when ``auto_paginate=True`` on
    :class:`~adcp.decisioning.platform.DecisioningCapabilities`.

    Short-circuits (returns *response* unchanged) when:

    * ``response.pagination`` is already populated — the adopter handled
      pagination natively; the framework must not overwrite it.
    * ``response.products`` is absent — unexpected shape; pass through and
      let wire validation surface the issue.

    Clamps ``max_results`` to ``[1, 100]`` before slicing.

    :param response: The adopter's full-list ``GetProductsResponse``.
    :param pagination: The wire ``Pagination`` request object
        (``max_results``, ``cursor``).
    :param query_hash: Filter fingerprint from :func:`_query_hash` on the
        original request. Used to anchor the cursor.
    :param secret: HMAC key override for testing and direct use. ``None`` uses
        :func:`_secret` (reads ``ADCP_PAGINATION_SECRET`` env var, falls back
        to the per-process random secret). The handler always passes ``None``
        — configure production secrets via the env var.
    :returns: A new ``GetProductsResponse`` with ``products`` sliced to the
        page and ``pagination`` populated, or the original *response* if the
        short-circuit fired.
    """
    # Short-circuit: adopter already populated pagination. Also short-circuits
    # when _invoke_platform_method returned a TaskHandoff projection dict
    # ({"task_id": ..., "status": "submitted"}) — dicts have no .products
    # attribute so the products-None branch below fires and passes through.
    if getattr(response, "pagination", None) is not None:
        return response

    products = getattr(response, "products", None)
    if products is None:
        return response

    # Decode cursor or start at offset 0.
    cursor_str = getattr(pagination, "cursor", None)
    if cursor_str:
        offset = _decode_cursor(cursor_str, query_hash, secret)
    else:
        offset = 0

    # Clamp max_results to wire schema bounds [1, 100].
    raw_max = getattr(pagination, "max_results", None)
    max_results = max(1, min(100, raw_max if raw_max is not None else 50))

    full_list = list(products)
    page = full_list[offset : offset + max_results]
    has_more = (offset + max_results) < len(full_list)
    next_cursor = (
        _encode_cursor(offset + max_results, query_hash, secret) if has_more else None
    )

    # Deferred: adcp.types imports adcp.decisioning.types; top-level import is circular.
    from adcp.types import PaginationResponse

    new_pagination = PaginationResponse(has_more=has_more, cursor=next_cursor)
    return response.model_copy(update={"products": page, "pagination": new_pagination})

Slice a full-list GetProductsResponse to the requested page.

Called by the handler post-adapter when auto_paginate=True on :class:~adcp.decisioning.platform.DecisioningCapabilities.

Short-circuits (returns response unchanged) when:

• response.pagination is already populated — the adopter handled pagination natively; the framework must not overwrite it.
• response.products is absent — unexpected shape; pass through and let wire validation surface the issue.

Clamps max_results to [1, 100] before slicing.

:param response: The adopter's full-list GetProductsResponse. :param pagination: The wire Pagination request object (max_results, cursor). :param query_hash: Filter fingerprint from :func:_query_hash on the original request. Used to anchor the cursor. :param secret: HMAC key override for testing and direct use. None uses :func:_secret (reads ADCP_PAGINATION_SECRET env var, falls back to the per-process random secret). The handler always passes None — configure production secrets via the env var. :returns: A new GetProductsResponse with products sliced to the page and pagination populated, or the original response if the short-circuit fired.