Module adcp.media_buy_actions

Buyer and seller helpers for proposal-bound MediaBuy change rights.

The helpers in this module deliberately keep three protocol surfaces separate:

• product allowed_actions are advisory possibilities;
• accepted proposal commercial_terms.change_terms are binding rights; and
• MediaBuy available_actions are the seller's current-state projection.

Legacy fields are readable, but never promoted into proposal authority. In particular, a 3.1 terms_ref is opaque unless this module itself emitted it as the compatibility projection of a known change_term_id.

Functions

def action_update_fields(version: str | None = None) ‑> Mapping[str, tuple[str, ...]]

Expand source code

@lru_cache(maxsize=16)
def action_update_fields(version: str | None = None) -> Mapping[str, tuple[str, ...]]:
    """Which ``update_media_buy`` request fields each action covers.

    AdCP 3.2.0-rc.3 splits this normative map across **two** ``enumMetadata``
    blocks and requires SDKs to merge them: the deprecated flat
    ``enums/media-buy-valid-action.json`` plus
    ``core/media-buy-available-action-id.json``, which carries the
    structured-only actions that were introduced after the flat surface was
    deprecated. Reading only the first silently omits
    ``update_media_buy_frequency_cap`` and its ``frequency_cap`` field, so a
    dispatcher built on it would route a valid rc.3 mutation nowhere.

    Read from the bundled schemas rather than transcribed, because the spec
    names this block -- not the task-reference table -- as the thing SDKs
    dispatch on. A hand-copied table is a second source of truth that drifts on
    the next release.

    ``version`` selects the bundle, so a client pinned to an older release
    dispatches on that release's map rather than on the SDK's default. Cached
    per version, because each is a pure function of an immutable bundle.

    Returns an empty mapping when the bundle predates the fields, so an older
    pin degrades rather than raising.
    """
    merged: dict[str, tuple[str, ...]] = {}
    for name in ("enums/media-buy-valid-action.json", "core/media-buy-available-action-id.json"):
        block = _schema_enum_metadata(name, version=version)
        for action, metadata in block.items():
            if action.startswith("$") or not isinstance(metadata, Mapping):
                continue
            fields = metadata.get("update_fields")
            if isinstance(fields, Sequence) and not isinstance(fields, str):
                merged[action] = tuple(str(item) for item in fields)
    return merged

Which update_media_buy request fields each action covers.

AdCP 3.2.0-rc.3 splits this normative map across two enumMetadata blocks and requires SDKs to merge them: the deprecated flat enums/media-buy-valid-action.json plus core/media-buy-available-action-id.json, which carries the structured-only actions that were introduced after the flat surface was deprecated. Reading only the first silently omits update_media_buy_frequency_cap and its frequency_cap field, so a dispatcher built on it would route a valid rc.3 mutation nowhere.

Read from the bundled schemas rather than transcribed, because the spec names this block – not the task-reference table – as the thing SDKs dispatch on. A hand-copied table is a second source of truth that drifts on the next release.

version selects the bundle, so a client pinned to an older release dispatches on that release's map rather than on the SDK's default. Cached per version, because each is a pure function of an immutable bundle.

Returns an empty mapping when the bundle predates the fields, so an older pin degrades rather than raising.

def assess_media_buy_action(action: str,
*,
product: Any | None = None,
proposal: Any | None = None,
media_buy: Any | None = None,
intent: ActionIntent | Mapping[str, Any] | None = None,
now: datetime | None = None) ‑> MediaBuyActionAssessment

Expand source code

def assess_media_buy_action(
    action: str,
    *,
    product: Any | None = None,
    proposal: Any | None = None,
    media_buy: Any | None = None,
    intent: ActionIntent | Mapping[str, Any] | None = None,
    now: datetime | None = None,
) -> MediaBuyActionAssessment:
    """Join product, accepted proposal, and live-buy action surfaces.

    Inputs may be generated Pydantic models or wire mappings.  Unknown future
    enum values and legacy coarse actions are retained as unknown evidence but
    are never treated as authority.
    """

    if _ACTION_RE.fullmatch(action) is None:
        raise MediaBuyActionError("invalid_action", "action")

    diagnostics: list[ActionDiagnostic] = []
    possible = _assess_product_support(action, product, diagnostics)
    change_terms_present, change_terms = _extract_change_terms(proposal, media_buy)
    term, term_diagnostics = _find_change_term(action, change_terms)
    diagnostics.extend(term_diagnostics)

    if not change_terms_present:
        promised = ActionKnowledge.unknown
        diagnostics.append(ActionDiagnostic(code=ActionDiagnosticCode.legacy_terms_unknown))
    elif term is None:
        promised = ActionKnowledge.no
    else:
        promised = ActionKnowledge.yes

    status_value = _string_field(media_buy, "status")
    wrong_status = False
    if term is not None and status_value is not None:
        allowed_statuses = _string_sequence(term.get("allowed_statuses"))
        wrong_status = status_value in _TERMINAL_STATUSES or (
            bool(allowed_statuses) and status_value not in allowed_statuses
        )

    live_state, live_action = _assess_live_action(action, media_buy, diagnostics)
    task: ActionTask | None = None
    mode: str | None = None
    change_term_id: str | None = None
    constraints: tuple[ConstraintCheck, ...] = ()

    if live_action is not None:
        task = _task_field(live_action) if "task" in live_action else route_media_buy_action(action)
        mode = _string_value(live_action.get("mode"))
        change_term_id = _string_value(live_action.get("change_term_id"))
        if term is not None and not _projection_matches_term(
            action, live_action, term, task, diagnostics
        ):
            live_state = ActionKnowledge.no
    elif term is not None:
        task = route_media_buy_action(action)
        mode = _string_value(term.get("service_mode"))
        change_term_id = _string_value(term.get("term_id"))

    if term is not None and intent is not None:
        parsed_intent = (
            intent if isinstance(intent, ActionIntent) else ActionIntent.model_validate(intent)
        )
        constraints = evaluate_action_constraints(
            action,
            term.get("constraints"),
            parsed_intent,
            now=now,
        )
        if any(check.outcome is ConstraintOutcome.violated for check in constraints):
            diagnostics.append(ActionDiagnostic(code=ActionDiagnosticCode.constraint_violated))
            live_state = ActionKnowledge.no

    integrity_failure = any(
        diagnostic.code
        in {
            ActionDiagnosticCode.alias_mismatch,
            ActionDiagnosticCode.duplicate_action,
            ActionDiagnosticCode.duplicate_term_id,
            ActionDiagnosticCode.invalid_projection,
            ActionDiagnosticCode.missing_change_term_link,
            ActionDiagnosticCode.mode_mismatch,
            ActionDiagnosticCode.route_mismatch,
            ActionDiagnosticCode.sla_mismatch,
        }
        for diagnostic in diagnostics
    )

    if action not in _CANONICAL_ACTIONS:
        diagnostics.append(ActionDiagnostic(code=ActionDiagnosticCode.unknown_action))
        final_status = ActionAvailabilityStatus.currently_unavailable
        live_state = ActionKnowledge.unknown
    elif integrity_failure:
        final_status = ActionAvailabilityStatus.currently_unavailable
        live_state = ActionKnowledge.no
    elif possible is ActionKnowledge.no:
        final_status = ActionAvailabilityStatus.unsupported_by_product
    elif promised is ActionKnowledge.no:
        final_status = ActionAvailabilityStatus.not_negotiated
    elif promised is ActionKnowledge.unknown:
        final_status = ActionAvailabilityStatus.legacy_unknown
    elif wrong_status:
        final_status = ActionAvailabilityStatus.wrong_status
        live_state = ActionKnowledge.no
    elif live_state is ActionKnowledge.yes:
        final_status = ActionAvailabilityStatus.available_now
    elif live_state is ActionKnowledge.unknown:
        final_status = ActionAvailabilityStatus.legacy_unknown
    else:
        final_status = ActionAvailabilityStatus.currently_unavailable

    return MediaBuyActionAssessment(
        action=action,
        status=final_status,
        possible=possible,
        promised=promised,
        available=live_state,
        task=task,
        mode=mode,
        change_term_id=change_term_id,
        async_processing=mode in {"seller_managed", "requires_approval"},
        constraints=constraints,
        diagnostics=tuple(_dedupe_diagnostics(diagnostics)),
    )

Join product, accepted proposal, and live-buy action surfaces.

Inputs may be generated Pydantic models or wire mappings. Unknown future enum values and legacy coarse actions are retained as unknown evidence but are never treated as authority.

def assess_update_media_buy_actions(patch: Any,
current_media_buy: Any,
*,
product: Any | None = None,
proposal: Any | None = None,
now: datetime | None = None) ‑> tuple[MediaBuyActionAssessment, ...]

Expand source code

def assess_update_media_buy_actions(
    patch: Any,
    current_media_buy: Any,
    *,
    product: Any | None = None,
    proposal: Any | None = None,
    now: datetime | None = None,
) -> tuple[MediaBuyActionAssessment, ...]:
    """Assess every logical action represented by an update compatibility patch."""

    assessments: list[MediaBuyActionAssessment] = []
    for mutation in decompose_update_media_buy(patch, current_media_buy):
        if mutation.action == UNKNOWN_UPDATE_ACTION:
            continue
        intent = _intent_from_mutation(mutation, current_media_buy, now=now)
        assessments.append(
            assess_media_buy_action(
                mutation.action,
                product=product,
                proposal=proposal,
                media_buy=current_media_buy,
                intent=intent,
                now=now,
            )
        )
    return tuple(assessments)

Assess every logical action represented by an update compatibility patch.

async def dispatch_media_buy_action(client: ActionDispatchClient,
assessment: MediaBuyActionAssessment,
request: BaseModel,
*,
options: Any | None = None) ‑> Any

Expand source code

async def dispatch_media_buy_action(
    client: ActionDispatchClient,
    assessment: MediaBuyActionAssessment,
    request: BaseModel,
    *,
    options: Any | None = None,
) -> Any:
    """Dispatch an already-assessed action through its canonical task.

    The caller still supplies the task-specific generated request model.  This
    helper owns only the action-to-task decision and refuses unavailable or
    unknown actions.  ``seller_managed`` uses the same ordinary async task
    lifecycle as every other mode; no internal approval workflow is inferred.
    """

    if assessment.status is not ActionAvailabilityStatus.available_now:
        raise MediaBuyActionError("action_not_available", "status")
    if assessment.task is None:
        raise MediaBuyActionError("unknown_action_route", "task")
    return await client.execute_task(assessment.task.value, request, options=options)

Dispatch an already-assessed action through its canonical task.

The caller still supplies the task-specific generated request model. This helper owns only the action-to-task decision and refuses unavailable or unknown actions. seller_managed uses the same ordinary async task lifecycle as every other mode; no internal approval workflow is inferred.

def evaluate_action_constraints(action: str,
constraints: Any | None,
intent: ActionIntent,
*,
now: datetime | None = None) ‑> tuple[ConstraintCheck, ...]

Expand source code

def evaluate_action_constraints(
    action: str,
    constraints: Any | None,
    intent: ActionIntent,
    *,
    now: datetime | None = None,
) -> tuple[ConstraintCheck, ...]:
    """Evaluate portable constraints without interpreting opaque conditions."""

    constraint = _as_mapping(constraints)
    if not constraint:
        return ()
    kind = _string_value(constraint.get("kind"))
    if kind is None or action not in _CONSTRAINT_ACTIONS.get(kind, frozenset()):
        return (
            ConstraintCheck(
                kind=kind or "unknown",
                constraint="kind",
                outcome=ConstraintOutcome.unknown,
                field=intent.field,
            ),
        )

    if kind == "budget":
        return _evaluate_budget_constraints(constraint, intent)
    if kind == "flight":
        return _evaluate_flight_constraints(constraint, intent, now=now)
    if kind == "package_count":
        return _evaluate_package_constraints(constraint, intent)
    if kind == "effective_timing":
        return _evaluate_effective_timing_constraints(constraint, intent, now=now)
    return ()

Evaluate portable constraints without interpreting opaque conditions.

def materialize_change_terms(product_actions: Iterable[Any],
selections: Iterable[ChangeTermSelection | Mapping[str, Any]]) ‑> tuple[MediaBuyChangeTerm, ...]

Expand source code

def materialize_change_terms(
    product_actions: Iterable[Any],
    selections: Iterable[ChangeTermSelection | Mapping[str, Any]],
) -> tuple[MediaBuyChangeTerm, ...]:
    """Materialize explicitly selected product templates as binding terms.

    Product declarations are never copied wholesale.  Every returned term
    requires a :class:`ChangeTermSelection`, and all portable bounds, SLA, and
    status scope are copied from (or narrowed relative to) that template.
    """

    product_index = _unique_action_index(product_actions, field="product_actions")
    parsed = [
        (
            value
            if isinstance(value, ChangeTermSelection)
            else ChangeTermSelection.model_validate(value)
        )
        for value in selections
    ]
    if len({selection.action for selection in parsed}) != len(parsed):
        raise MediaBuyActionError("duplicate_action", "selections")
    if len({selection.term_id for selection in parsed}) != len(parsed):
        raise MediaBuyActionError("duplicate_term_id", "selections")

    terms: list[MediaBuyChangeTerm] = []
    for index, selection in enumerate(parsed):
        product_action = product_index.get(selection.action)
        if product_action is None:
            raise MediaBuyActionError("action_not_advertised", f"selections[{index}].action")
        if selection.action not in _CANONICAL_ACTIONS:
            raise MediaBuyActionError("unknown_action", f"selections[{index}].action")

        modes = _string_sequence(product_action.get("modes"))
        mode = selection.service_mode
        if mode is None:
            if len(modes) != 1:
                raise MediaBuyActionError(
                    "service_mode_required", f"selections[{index}].service_mode"
                )
            mode = modes[0]
        if mode not in modes:
            raise MediaBuyActionError(
                "service_mode_not_advertised", f"selections[{index}].service_mode"
            )

        product_statuses = _string_sequence(product_action.get("allowed_statuses"))
        selected_statuses = selection.allowed_statuses
        if (
            selected_statuses is not None
            and product_statuses
            and not set(selected_statuses) <= set(product_statuses)
        ):
            raise MediaBuyActionError(
                "status_scope_expanded", f"selections[{index}].allowed_statuses"
            )
        statuses = selected_statuses or (tuple(product_statuses) if product_statuses else None)

        constraint = _as_mapping(product_action.get("constraints")) or None
        if constraint is not None:
            kind = _string_value(constraint.get("kind"))
            if selection.action not in _CONSTRAINT_ACTIONS.get(kind or "", frozenset()):
                raise MediaBuyActionError(
                    "constraint_action_mismatch",
                    f"product_actions[{index}].constraints",
                )

        term_payload: dict[str, Any] = {
            "term_id": selection.term_id,
            "action": selection.action,
            "service_mode": mode,
        }
        if statuses is not None:
            term_payload["allowed_statuses"] = list(statuses)
        sla = _as_mapping(product_action.get("sla"))
        if sla:
            term_payload["processing_sla"] = sla
        if constraint is not None:
            term_payload["constraints"] = constraint
        if selection.conditions is not None:
            term_payload["conditions"] = list(selection.conditions)
        terms_ref = selection.terms_ref or _string_value(product_action.get("terms_ref"))
        if terms_ref is not None:
            term_payload["terms_ref"] = terms_ref
        if selection.description is not None:
            term_payload["description"] = selection.description
        terms.append(MediaBuyChangeTerm.model_validate(term_payload))
    return tuple(terms)

Materialize explicitly selected product templates as binding terms.

Product declarations are never copied wholesale. Every returned term requires a :class:ChangeTermSelection, and all portable bounds, SLA, and status scope are copied from (or narrowed relative to) that template.

def project_available_actions(change_terms: Iterable[Any] | None,
status: str,
*,
product_actions: Iterable[Any] | None = None,
authorized_actions: Iterable[str] | None = None,
delegated_actions: Iterable[str] | None = None,
policy_actions: Iterable[str] | None = None,
resolved_conditions: Mapping[str, bool] | None = None,
protocol_version: str = '3.2') ‑> MediaBuyActionProjection

Expand source code

def project_available_actions(
    change_terms: Iterable[Any] | None,
    status: str,
    *,
    product_actions: Iterable[Any] | None = None,
    authorized_actions: Iterable[str] | None = None,
    delegated_actions: Iterable[str] | None = None,
    policy_actions: Iterable[str] | None = None,
    resolved_conditions: Mapping[str, bool] | None = None,
    protocol_version: str = "3.2",
) -> MediaBuyActionProjection:
    """Derive a fail-closed seller ``available_actions`` projection.

    ``None`` for an optional narrowing set means that gate has already been
    satisfied or is not applicable.  An explicit empty set denies every
    action.  Conditions must be explicitly resolved ``True``; absent, false,
    or unknown conditions omit the action.
    """

    diagnostics: list[ActionDiagnostic] = []
    if change_terms is None:
        return MediaBuyActionProjection(
            diagnostics=(ActionDiagnostic(code=ActionDiagnosticCode.legacy_terms_unknown),)
        )
    if status not in _NON_TERMINAL_STATUSES:
        return MediaBuyActionProjection()

    terms: list[Mapping[str, Any]] = []
    for index, raw_term in enumerate(change_terms):
        try:
            parsed_term = MediaBuyChangeTerm.model_validate(raw_term)
        except ValidationError:
            diagnostics.append(
                ActionDiagnostic(
                    code=ActionDiagnosticCode.invalid_projection,
                    field=f"change_terms[{index}]",
                )
            )
            continue
        terms.append(parsed_term.model_dump(mode="python", by_alias=True, exclude_unset=True))
    seen_actions: set[str] = set()
    seen_ids: set[str] = set()
    product_index = (
        _unique_action_index(product_actions, field="product_actions")
        if product_actions is not None
        else None
    )
    gates = [
        set(values) if values is not None else None
        for values in (authorized_actions, delegated_actions, policy_actions)
    ]
    supports_link = _supports_change_term_id(protocol_version)
    supports_task = _supports_action_task(protocol_version)
    actions: list[ProjectedMediaBuyAction] = []

    for index, term in enumerate(terms):
        action = _string_value(term.get("action"))
        term_id = _string_value(term.get("term_id"))
        if action is None or term_id is None or _TERM_ID_RE.fullmatch(term_id) is None:
            diagnostics.append(
                ActionDiagnostic(
                    code=ActionDiagnosticCode.invalid_projection,
                    field=f"change_terms[{index}]",
                )
            )
            continue
        if action in seen_actions:
            diagnostics.append(_diagnostic(ActionDiagnosticCode.duplicate_action, detail=action))
            continue
        if term_id in seen_ids:
            diagnostics.append(_diagnostic(ActionDiagnosticCode.duplicate_term_id, detail=term_id))
            continue
        seen_actions.add(action)
        seen_ids.add(term_id)

        task = route_media_buy_action(action)
        if task is None:
            diagnostics.append(_diagnostic(ActionDiagnosticCode.unknown_action, detail=action))
            continue
        allowed_statuses = _string_sequence(term.get("allowed_statuses"))
        if allowed_statuses and status not in allowed_statuses:
            continue
        if any(gate is not None and action not in gate for gate in gates):
            continue

        product_action = product_index.get(action) if product_index is not None else None
        if product_index is not None and product_action is None:
            continue
        mode = _string_value(term.get("service_mode"))
        if mode is None:
            diagnostics.append(
                ActionDiagnostic(
                    code=ActionDiagnosticCode.invalid_projection,
                    field=f"change_terms[{index}].service_mode",
                )
            )
            continue
        if product_action is not None:
            if mode not in _string_sequence(product_action.get("modes")):
                continue
            product_statuses = _string_sequence(product_action.get("allowed_statuses"))
            if product_statuses and status not in product_statuses:
                continue

        conditions = _string_sequence(term.get("conditions"))
        if conditions and (
            resolved_conditions is None
            or any(resolved_conditions.get(condition) is not True for condition in conditions)
        ):
            diagnostics.append(
                _diagnostic(ActionDiagnosticCode.condition_unresolved, detail=action)
            )
            continue

        constraint = _as_mapping(term.get("constraints"))
        if constraint:
            kind = _string_value(constraint.get("kind"))
            if action not in _CONSTRAINT_ACTIONS.get(kind or "", frozenset()):
                diagnostics.append(
                    _diagnostic(ActionDiagnosticCode.invalid_projection, detail=action)
                )
                continue

        wire_mode = "requires_approval" if not supports_link and mode == "seller_managed" else mode
        action_payload: dict[str, Any] = {
            "action": action,
            "mode": wire_mode,
            "sla": _as_mapping(term.get("processing_sla")) or None,
        }
        if supports_task:
            action_payload["task"] = task
        if supports_link:
            action_payload["change_term_id"] = term_id
        else:
            # This is an explicit adapter-generated alias.  Arbitrary inbound
            # 3.1 terms_ref values are never interpreted this way.
            action_payload["terms_ref"] = term_id
        actions.append(ProjectedMediaBuyAction.model_validate(action_payload))

    return MediaBuyActionProjection(
        actions=tuple(actions), diagnostics=tuple(_dedupe_diagnostics(diagnostics))
    )

Derive a fail-closed seller available_actions projection.

None for an optional narrowing set means that gate has already been satisfied or is not applicable. An explicit empty set denies every action. Conditions must be explicitly resolved True; absent, false, or unknown conditions omit the action.

def route_media_buy_action(action: str, *, version: str | None = None) ‑> ActionTask | None

Expand source code

def route_media_buy_action(action: str, *, version: str | None = None) -> ActionTask | None:
    """Return the canonical compact task for an in-envelope action.

    Flight and package-addition changes require proposal refinement; creative
    lifecycle changes use ``sync_creatives``; the remaining accepted controls
    use ``control_media_buy``.

    A structured-only action the tables above have not been taught still routes
    when the bundle's merged ``enumMetadata`` says it mutates
    ``update_media_buy`` fields. The spec adds these additively and tells SDKs
    to dispatch on that block, so consulting it keeps a newer bundle working
    instead of failing closed on a name the tables predate -- which is the bug
    ``update_media_buy_frequency_cap`` exposed. Anything the bundle does not
    describe either still fails closed with ``None`` so callers do not guess a
    route.
    """

    if action in _CONTROL_ACTIONS:
        return ActionTask.control_media_buy
    if action in _REFINEMENT_ACTIONS:
        return ActionTask.refine_proposals
    if action in _CREATIVE_ACTIONS:
        return ActionTask.sync_creatives
    if action_update_fields(version).get(action):
        return ActionTask.control_media_buy
    return None

Return the canonical compact task for an in-envelope action.

Flight and package-addition changes require proposal refinement; creative lifecycle changes use sync_creatives; the remaining accepted controls use control_media_buy.

A structured-only action the tables above have not been taught still routes when the bundle's merged enumMetadata says it mutates update_media_buy fields. The spec adds these additively and tells SDKs to dispatch on that block, so consulting it keeps a newer bundle working instead of failing closed on a name the tables predate – which is the bug update_media_buy_frequency_cap exposed. Anything the bundle does not describe either still fails closed with None so callers do not guess a route.

Classes

class ActionAvailabilityStatus (*args, **kwds)

Expand source code

class ActionAvailabilityStatus(StrEnum):
    """Normalized buyer-facing action assessment result."""

    available_now = "available_now"
    wrong_status = "wrong_status"
    not_negotiated = "not_negotiated"
    unsupported_by_product = "unsupported_by_product"
    currently_unavailable = "currently_unavailable"
    legacy_unknown = "legacy_unknown"

Normalized buyer-facing action assessment result.

Ancestors

• enum.StrEnum
• builtins.str
• enum.ReprEnum
• enum.Enum

Class variables

var available_now

var currently_unavailable

var legacy_unknown

var not_negotiated

var unsupported_by_product

var wrong_status

class ActionDiagnostic (**data: Any)

Expand source code

class ActionDiagnostic(BaseModel):
    """Bounded diagnostic suitable for UI, logs, and agent reasoning."""

    model_config = ConfigDict(extra="forbid", frozen=True)

    code: ActionDiagnosticCode
    field: str | None = Field(default=None, max_length=128, pattern=r"^[A-Za-z0-9_.\[\]-]+$")
    detail: str | None = Field(
        default=None,
        max_length=128,
        pattern=r"^[A-Za-z][A-Za-z0-9_.:-]{0,127}$",
    )

Bounded diagnostic suitable for UI, logs, and agent reasoning.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var code : ActionDiagnosticCode

var detail : str | None

var field : str | None

var model_config

class ActionDiagnosticCode (*args, **kwds)

Expand source code

class ActionDiagnosticCode(StrEnum):
    """Stable machine-readable explanations emitted by the resolver."""

    alias_mismatch = "alias_mismatch"
    condition_unresolved = "condition_unresolved"
    constraint_violated = "constraint_violated"
    duplicate_action = "duplicate_action"
    duplicate_term_id = "duplicate_term_id"
    invalid_projection = "invalid_projection"
    legacy_coarse_action = "legacy_coarse_action"
    legacy_terms_unknown = "legacy_terms_unknown"
    missing_change_term_link = "missing_change_term_link"
    mode_mismatch = "mode_mismatch"
    route_mismatch = "route_mismatch"
    sla_mismatch = "sla_mismatch"
    unknown_action = "unknown_action"

Stable machine-readable explanations emitted by the resolver.

Ancestors

• enum.StrEnum
• builtins.str
• enum.ReprEnum
• enum.Enum

Class variables

var alias_mismatch

var condition_unresolved

var constraint_violated

var duplicate_action

var duplicate_term_id

var invalid_projection

var legacy_coarse_action

var legacy_terms_unknown

var missing_change_term_link

var mode_mismatch

var route_mismatch

var sla_mismatch

var unknown_action

class ActionDispatchClient (*args, **kwargs)

Expand source code

class ActionDispatchClient(Protocol):
    """Minimal client surface consumed by :func:`dispatch_media_buy_action`."""

    async def execute_task(
        self,
        task_name: str,
        request: BaseModel,
        *,
        options: Any | None = None,
    ) -> Any:
        raise NotImplementedError

Minimal client surface consumed by :func:dispatch_media_buy_action().

Ancestors

• typing.Protocol
• typing.Generic

Methods

async def execute_task(self, task_name: str, request: BaseModel, *, options: Any | None = None) ‑> Any

Expand source code

async def execute_task(
    self,
    task_name: str,
    request: BaseModel,
    *,
    options: Any | None = None,
) -> Any:
    raise NotImplementedError

class ActionIntent (**data: Any)

Expand source code

class ActionIntent(BaseModel):
    """Portable current/result state used to preflight typed constraints.

    Callers may provide this directly.  :func:`assess_update_media_buy_actions`
    derives the same fields from an update patch when enough current state is
    available.
    """

    model_config = ConfigDict(extra="forbid", frozen=True)

    current_amount: Decimal | None = None
    result_amount: Decimal | None = None
    currency: str | None = Field(default=None, pattern=r"^[A-Z]{3}$")
    current_time: datetime | None = None
    result_time: datetime | None = None
    effective_at: datetime | None = None
    additions: int | None = Field(default=None, ge=0)
    removals: int | None = Field(default=None, ge=0)
    current_package_count: int | None = Field(default=None, ge=0)
    result_package_count: int | None = Field(default=None, ge=0)
    field: str | None = Field(default=None, max_length=128, pattern=r"^[A-Za-z0-9_.\[\]-]+$")

Portable current/result state used to preflight typed constraints.

Callers may provide this directly. :func:assess_update_media_buy_actions() derives the same fields from an update patch when enough current state is available.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var additions : int | None

var currency : str | None

var current_amount : decimal.Decimal | None

var current_package_count : int | None

var current_time : datetime.datetime | None

var effective_at : datetime.datetime | None

var field : str | None

var model_config

var removals : int | None

var result_amount : decimal.Decimal | None

var result_package_count : int | None

var result_time : datetime.datetime | None

class ActionKnowledge (*args, **kwds)

Expand source code

class ActionKnowledge(StrEnum):
    """Whether a protocol surface proves, disproves, or cannot answer a fact."""

    yes = "yes"
    no = "no"
    unknown = "unknown"

Whether a protocol surface proves, disproves, or cannot answer a fact.

Ancestors

• enum.StrEnum
• builtins.str
• enum.ReprEnum
• enum.Enum

Class variables

var no

var unknown

var yes

class ActionTask (*args, **kwds)

Expand source code

class ActionTask(StrEnum):
    """Compact-lifecycle task used to exercise an action."""

    control_media_buy = "control_media_buy"
    refine_proposals = "refine_proposals"
    sync_creatives = "sync_creatives"

Compact-lifecycle task used to exercise an action.

Ancestors

• enum.StrEnum
• builtins.str
• enum.ReprEnum
• enum.Enum

Class variables

var control_media_buy

var refine_proposals

var sync_creatives

class ChangeTermSelection (**data: Any)

Expand source code

class ChangeTermSelection(BaseModel):
    """A seller's explicit decision to bind one advertised product action."""

    model_config = ConfigDict(extra="forbid", frozen=True)

    action: str = Field(max_length=128, pattern=r"^[A-Za-z][A-Za-z0-9_.:-]{0,127}$")
    term_id: str = Field(max_length=255, pattern=r"^[A-Za-z0-9_.:-]+$")
    service_mode: str | None = Field(
        default=None, max_length=64, pattern=r"^[A-Za-z][A-Za-z0-9_.:-]*$"
    )
    allowed_statuses: tuple[str, ...] | None = None
    conditions: tuple[str, ...] | None = None
    terms_ref: str | None = Field(default=None, min_length=1, max_length=1000)
    description: str | None = Field(default=None, min_length=1, max_length=1000)

    @model_validator(mode="after")
    def _validate_values(self) -> ChangeTermSelection:
        if self.allowed_statuses is not None:
            if not self.allowed_statuses or len(set(self.allowed_statuses)) != len(
                self.allowed_statuses
            ):
                raise ValueError("allowed_statuses must be non-empty and unique")
            if any(status not in _NON_TERMINAL_STATUSES for status in self.allowed_statuses):
                raise ValueError("allowed_statuses may contain only non-terminal statuses")
        if self.conditions is not None:
            if not self.conditions or len(set(self.conditions)) != len(self.conditions):
                raise ValueError("conditions must be non-empty and unique")
            if any(
                re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.:-]{0,199}", condition) is None
                for condition in self.conditions
            ):
                raise ValueError("condition identifiers must use the protocol token grammar")
        return self

A seller's explicit decision to bind one advertised product action.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var action : str

var allowed_statuses : tuple[str, ...] | None

var conditions : tuple[str, ...] | None

var description : str | None

var model_config

var service_mode : str | None

var term_id : str

var terms_ref : str | None

class ConstraintCheck (**data: Any)

Expand source code

class ConstraintCheck(BaseModel):
    """Evaluation of one portable constraint field."""

    model_config = ConfigDict(extra="forbid", frozen=True)

    kind: str = Field(max_length=32, pattern=r"^[a-z_]+$")
    constraint: str = Field(max_length=64, pattern=r"^[a-z_]+$")
    outcome: ConstraintOutcome
    field: str | None = Field(default=None, max_length=128, pattern=r"^[A-Za-z0-9_.\[\]-]+$")

Evaluation of one portable constraint field.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var constraint : str

var field : str | None

var kind : str

var model_config

var outcome : ConstraintOutcome

class ConstraintOutcome (*args, **kwds)

Expand source code

class ConstraintOutcome(StrEnum):
    """Result of preflighting one portable change-term constraint."""

    satisfied = "satisfied"
    violated = "violated"
    unknown = "unknown"

Result of preflighting one portable change-term constraint.

Ancestors

• enum.StrEnum
• builtins.str
• enum.ReprEnum
• enum.Enum

Class variables

var satisfied

var unknown

var violated

class MediaBuyActionAssessment (**data: Any)

Expand source code

class MediaBuyActionAssessment(BaseModel):
    """The joined possible/promised/current answer for one action."""

    model_config = ConfigDict(extra="forbid", frozen=True)

    action: str = Field(max_length=128, pattern=r"^[A-Za-z][A-Za-z0-9_.:-]{0,127}$")
    status: ActionAvailabilityStatus
    possible: ActionKnowledge
    promised: ActionKnowledge
    available: ActionKnowledge
    task: ActionTask | None = None
    mode: str | None = Field(default=None, max_length=64, pattern=r"^[A-Za-z][A-Za-z0-9_.:-]*$")
    change_term_id: str | None = Field(default=None, max_length=255, pattern=r"^[A-Za-z0-9_.:-]+$")
    async_processing: bool = False
    constraints: tuple[ConstraintCheck, ...] = ()
    diagnostics: tuple[ActionDiagnostic, ...] = ()

The joined possible/promised/current answer for one action.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var action : str

var async_processing : bool

var available : ActionKnowledge

var change_term_id : str | None

var constraints : tuple[ConstraintCheck, ...]

var diagnostics : tuple[ActionDiagnostic, ...]

var mode : str | None

var model_config

var possible : ActionKnowledge

var promised : ActionKnowledge

var status : ActionAvailabilityStatus

var task : ActionTask | None

class MediaBuyActionError (code: str, field: str | None = None)

Expand source code

class MediaBuyActionError(ValueError):
    """A safe local validation error for invalid seller helper inputs."""

    def __init__(self, code: str, field: str | None = None) -> None:
        self.code = code
        self.field = field
        suffix = f" at {field}" if field is not None else ""
        super().__init__(f"media-buy action validation failed: {code}{suffix}")

A safe local validation error for invalid seller helper inputs.

Ancestors

• builtins.ValueError
• builtins.Exception
• builtins.BaseException

class MediaBuyActionProjection (**data: Any)

Expand source code

class MediaBuyActionProjection(BaseModel):
    """Seller current-state projection plus any fail-closed omissions."""

    model_config = ConfigDict(extra="forbid", frozen=True)

    actions: tuple[ProjectedMediaBuyAction, ...] = ()
    diagnostics: tuple[ActionDiagnostic, ...] = ()

    def to_wire(self) -> list[dict[str, object]]:
        """Return the JSON-ready ``available_actions`` array."""

        return [action.model_dump(mode="json", exclude_none=True) for action in self.actions]

Seller current-state projection plus any fail-closed omissions.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var actions : tuple[ProjectedMediaBuyAction, ...]

var diagnostics : tuple[ActionDiagnostic, ...]

var model_config

Methods

def to_wire(self) ‑> list[dict[str, object]]

Expand source code

def to_wire(self) -> list[dict[str, object]]:
    """Return the JSON-ready ``available_actions`` array."""

    return [action.model_dump(mode="json", exclude_none=True) for action in self.actions]

Return the JSON-ready available_actions array.

class ProjectedMediaBuyAction (**data: Any)

Expand source code

class ProjectedMediaBuyAction(BaseModel):
    """Version-adaptable seller projection for one currently available action."""

    model_config = ConfigDict(extra="forbid", frozen=True)

    action: str = Field(max_length=128, pattern=r"^[A-Za-z][A-Za-z0-9_.:-]{0,127}$")
    mode: str = Field(max_length=64, pattern=r"^[A-Za-z][A-Za-z0-9_.:-]*$")
    task: ActionTask | None = None
    sla: dict[str, object] | None = None
    change_term_id: str | None = Field(default=None, max_length=255, pattern=r"^[A-Za-z0-9_.:-]+$")
    terms_ref: str | None = Field(default=None, max_length=1000)

Version-adaptable seller projection for one currently available action.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var action : str

var change_term_id : str | None

var mode : str

var model_config

var sla : dict[str, object] | None

var task : ActionTask | None

var terms_ref : str | None