Module `adcp.compat.legacy.v2_5.get_products`

v2.5 → v3 adapter for get_products.

This is the inverse of the JS SDK's adaptGetProductsRequestForV2 / normalizeGetProductsResponse helpers — the JS direction is client side (v3 client → v2 server). Server side flips both arrows:

Request (v2.5 → v3): the buyer sent a v2.5 shape; rewrite to v3 so the handler's typed code sees the canonical model.
Response (v3 → v2.5): the handler produced a v3 shape; rewrite to the v2.5 shape the buyer expects on the wire.

Field deltas the adapter handles:

brand_manifest (v2.5 URL string) ↔ brand (v3 BrandReference). v2.5 sent brand_manifest: "https://example.com"; v3 expects brand: {domain: "example.com"}.
promoted_offerings (v2.5 nested object) ↔ catalog (v3 discriminated union). v2.5's product-selectors / offerings nesting collapses to v3's catalog.type = 'product' | 'offering' shape.
Channels. v2.5 used coarser buckets (video, audio, native, retail); v3 splits them (video → olv+ctv, audio → streaming_audio, etc.). The buyer-direction mapping fans out; the response-direction mapping collapses — both are inherently lossy when traffic mixes the buckets. The translation defers ambiguity to the canonical mappings documented in the JS SDK and falls through unknown channels unchanged.
Pricing options. Per pricing option: rate ↔ fixed_price (and is_fixed discriminator gone in v3), price_guidance.floor ↔ floor_price (floor moves out of guidance). Percentile fields (p25/p50/p75/p90) stay in price_guidance in both versions.

Direct port (with direction inverted) of src/lib/utils/pricing-adapter.ts.

Functions

def adapt_request(payload: dict[str, Any]) ‑> dict[str, typing.Any]

Expand source code

def adapt_request(payload: dict[str, Any]) -> dict[str, Any]:
    """Translate a v2.5 ``get_products`` request to v3 shape."""
    out = dict(payload)

    # brand_manifest (v2.5 URL string or inline BrandManifest object) →
    # brand.domain (v3 BrandReference).
    # v2.5 brand-manifest-ref.json is a oneOf: either a URL string or an
    # inline BrandManifest object. The inline schema has url as optional;
    # when absent there is no derivable hostname, so we skip brand and let
    # v3 validation handle the missing field.
    brand_manifest = out.pop("brand_manifest", None)
    if isinstance(brand_manifest, str) and brand_manifest and "brand" not in out:
        domain = extract_brand_domain(brand_manifest)
        warn_brand_manifest_path_lossy(brand_manifest, domain)
        out["brand"] = {"domain": domain}
    elif isinstance(brand_manifest, dict) and "brand" not in out:
        url = brand_manifest.get("url")
        if isinstance(url, str) and url.strip():
            domain = extract_brand_domain(url)
            warn_brand_manifest_path_lossy(url, domain)
            out["brand"] = {"domain": domain}

    # promoted_offerings → catalog
    promoted = out.pop("promoted_offerings", None)
    if isinstance(promoted, dict) and "catalog" not in out:
        product_selectors = promoted.get("product_selectors")
        if isinstance(product_selectors, dict):
            catalog: dict[str, Any] = {"type": "product"}
            ps = product_selectors
            if "manifest_gtins" in ps:
                catalog["gtins"] = ps["manifest_gtins"]
            if "manifest_skus" in ps:
                catalog["ids"] = ps["manifest_skus"]
            if "manifest_tags" in ps:
                catalog["tags"] = ps["manifest_tags"]
            if "manifest_category" in ps:
                catalog["category"] = ps["manifest_category"]
            if "manifest_query" in ps:
                catalog["query"] = ps["manifest_query"]
            out["catalog"] = catalog
        elif "offerings" in promoted:
            out["catalog"] = {"type": "offering", "items": promoted["offerings"]}

    # Channel-bucket fan-out in filters.
    filters = out.get("filters")
    if isinstance(filters, dict) and isinstance(filters.get("channels"), list):
        expanded: list[str] = []
        seen: set[str] = set()
        for ch in filters["channels"]:
            if not isinstance(ch, str):
                continue
            mapped = _V2_TO_V3_CHANNEL.get(ch, (ch,))
            for slug in mapped:
                if slug not in seen:
                    seen.add(slug)
                    expanded.append(slug)
        out["filters"] = {**filters, "channels": expanded}

    return out

Translate a v2.5 get_products request to v3 shape.

def is_legacy_shape(payload: dict[str, Any]) ‑> bool

Expand source code

def is_legacy_shape(payload: dict[str, Any]) -> bool:
    """v2.5 ``get_products`` carries either ``brand_manifest`` (URL
    string or inline object — v3 has no ``brand_manifest`` key) or
    ``promoted_offerings`` (nested object replaced by ``catalog`` in v3).
    Either is a strong signal."""
    return "brand_manifest" in payload or "promoted_offerings" in payload

v2.5 get_products carries either brand_manifest (URL string or inline object — v3 has no brand_manifest key) or promoted_offerings (nested object replaced by catalog in v3). Either is a strong signal.

def normalize_response(response: dict[str, Any]) ‑> dict[str, typing.Any]

Expand source code

def normalize_response(response: dict[str, Any]) -> dict[str, Any]:
    """Translate a v3 ``get_products`` response back to v2.5 shape.

    Mirrors ``adaptGetProductsResponseForV2`` (the inverse of the JS
    ``normalizeGetProductsResponse``). For each product:

    * Collapse v3 channel slugs to v2.5 buckets (lossy — ``olv`` and
      ``ctv`` both collapse to ``video``).
    * Rewrite each pricing option's ``fixed_price`` / ``floor_price``
      back to ``rate`` / ``price_guidance.floor`` with the ``is_fixed``
      discriminator restored.
    """
    products = response.get("products")
    if not isinstance(products, list):
        return response
    return {
        **response,
        "products": [
            (
                _normalize_product_pricing_v3_to_v2(_normalize_product_channels(p))
                if isinstance(p, dict)
                else p
            )
            for p in products
        ],
    }

Translate a v3 get_products response back to v2.5 shape.

Mirrors adaptGetProductsResponseForV2 (the inverse of the JS normalizeGetProductsResponse). For each product:

Collapse v3 channel slugs to v2.5 buckets (lossy — olv and ctv both collapse to video).
Rewrite each pricing option's fixed_price / floor_price back to rate / price_guidance.floor with the is_fixed discriminator restored.