Module adcp.decisioning.state_machines

Canonical state-machine helpers for AdCP lifecycle objects.

The AdCP spec defines a finite-state graph for media buys and creatives; adopters who hand-roll lifecycle methods (update_media_buy, sync_creatives, etc.) repeatedly re-implement the same legality checks. These helpers project the spec state graph as a single source of truth and raise a spec-conformant INVALID_STATE error when an adopter tries an illegal transition.

Use in update_media_buy / lifecycle-changing handlers to refuse non-monotonic state changes (e.g., active → pending_creatives). The helpers are deliberately small — adopters who need richer guards (e.g., role-based authorization on transitions) compose their own checks alongside these.

State enum values track the AdCP spec at schemas/cache/enums/media-buy-status.json and schemas/cache/enums/creative-status.json.

Global variables

var CREATIVE_ASSET_TRANSITIONS : Mapping[str, frozenset[str]]

Legal creative-asset state transitions per the AdCP spec. Maps from_state → set of legal to_state values. Terminal states map to an empty frozenset.

var MEDIA_BUY_TRANSITIONS : Mapping[str, frozenset[str]]

Legal media-buy state transitions per the AdCP spec. Maps from_state → set of legal to_state values. Terminal states map to an empty frozenset.

Functions

def assert_creative_transition(from_state: str, to_state: str, *, creative_id: str | None = None) ‑> None

Expand source code

def assert_creative_transition(
    from_state: str,
    to_state: str,
    *,
    creative_id: str | None = None,
) -> None:
    """Raise ``INVALID_STATE`` on illegal creative-asset transitions.

    Use inside ``sync_creatives`` / creative-approval handlers to refuse
    non-monotonic state changes (e.g., ``archived`` → ``approved``).

    :param from_state: Current creative status.
    :param to_state: Requested next status.
    :param creative_id: Optional id, surfaced in ``error.details`` for
        buyer-side debugging.
    :raises AdcpError: with ``code='INVALID_STATE'`` and
        ``recovery='correctable'`` per the spec when the transition is
        not in :data:`CREATIVE_ASSET_TRANSITIONS`.
    """
    _assert_transition(
        from_state,
        to_state,
        graph=CREATIVE_ASSET_TRANSITIONS,
        resource_kind="creative",
        resource_id=creative_id,
        id_field="creative_id",
    )

Raise INVALID_STATE on illegal creative-asset transitions.

Use inside sync_creatives / creative-approval handlers to refuse non-monotonic state changes (e.g., archived → approved).

:param from_state: Current creative status. :param to_state: Requested next status. :param creative_id: Optional id, surfaced in error.details for buyer-side debugging. :raises AdcpError: with code='INVALID_STATE' and recovery='correctable' per the spec when the transition is not in :data:CREATIVE_ASSET_TRANSITIONS.

def assert_media_buy_transition(from_state: str, to_state: str, *, media_buy_id: str | None = None) ‑> None

Expand source code

def assert_media_buy_transition(
    from_state: str,
    to_state: str,
    *,
    media_buy_id: str | None = None,
) -> None:
    """Raise ``INVALID_STATE`` on illegal media-buy transitions.

    Use inside ``update_media_buy`` / ``cancel_media_buy`` /
    lifecycle-changing handlers to refuse non-monotonic state changes
    (e.g., ``active`` → ``pending_creatives``).

    :param from_state: Current media-buy status.
    :param to_state: Requested next status.
    :param media_buy_id: Optional id, surfaced in ``error.details`` for
        buyer-side debugging.
    :raises AdcpError: with ``code='INVALID_STATE'`` and
        ``recovery='correctable'`` per the spec when the transition is
        not in :data:`MEDIA_BUY_TRANSITIONS`.
    """
    _assert_transition(
        from_state,
        to_state,
        graph=MEDIA_BUY_TRANSITIONS,
        resource_kind="media buy",
        resource_id=media_buy_id,
        id_field="media_buy_id",
    )

Raise INVALID_STATE on illegal media-buy transitions.

Use inside update_media_buy / cancel_media_buy / lifecycle-changing handlers to refuse non-monotonic state changes (e.g., active → pending_creatives).

:param from_state: Current media-buy status. :param to_state: Requested next status. :param media_buy_id: Optional id, surfaced in error.details for buyer-side debugging. :raises AdcpError: with code='INVALID_STATE' and recovery='correctable' per the spec when the transition is not in :data:MEDIA_BUY_TRANSITIONS.