Module adcp.decisioning.update_media_buy
Helpers for interpreting update_media_buy patch requests.
Functions
def decompose_update_media_buy(patch: Any, current_media_buy: Any | None = None) ‑> list[UpdateMediaBuyMutation]-
Expand source code
def decompose_update_media_buy( patch: Any, current_media_buy: Any | None = None, ) -> list[UpdateMediaBuyMutation]: """Split an ``update_media_buy`` patch into ordered logical mutations. ``patch`` may be a generated Pydantic ``UpdateMediaBuyRequest`` or a plain mapping. ``current_media_buy`` is optional; when supplied, the helper can promote coarse mutations into specific actions such as ``increase_budget``, ``decrease_budget``, ``extend_flight``, ``shorten_flight``, and ``reallocate_budget``. """ patch_dict = _to_plain_mapping(patch) current_dict = _to_plain_mapping(current_media_buy) if current_media_buy is not None else {} current_packages = _index_packages(current_dict.get("packages")) mutations: list[UpdateMediaBuyMutation] = [] if patch_dict.get("paused") is True: mutations.append( _mutation( "pause", ("paused",), before=_current_paused(current_dict), after=True, ) ) elif patch_dict.get("paused") is False: mutations.append( _mutation( "resume", ("paused",), before=_current_paused(current_dict), after=False, ) ) if patch_dict.get("canceled") is True: fields = ["canceled"] after: dict[str, Any] = {"canceled": True} if "cancellation_reason" in patch_dict: fields.append("cancellation_reason") after["cancellation_reason"] = patch_dict["cancellation_reason"] mutations.append( _mutation( "cancel", tuple(fields), before=_current_status(current_dict), after=after, raw=after, ) ) elif "cancellation_reason" in patch_dict: mutations.append( _unknown_mutation( ("cancellation_reason",), after=patch_dict["cancellation_reason"], ) ) date_fields = tuple( field_name for field_name in ("start_time", "end_time") if field_name in patch_dict ) if date_fields: before = { field_name: current_dict[field_name] for field_name in date_fields if field_name in current_dict } after = {field_name: patch_dict[field_name] for field_name in date_fields} action, resolution = _date_action(date_fields, before, after) mutations.append( _mutation( action, date_fields, before=before or None, after=after, raw=after, resolution=resolution, ) ) if "new_packages" in patch_dict: mutations.append( _mutation( "add_packages", ("new_packages",), after=patch_dict["new_packages"], raw=patch_dict["new_packages"], ) ) packages = _as_sequence_of_mappings(patch_dict.get("packages")) reallocation = _package_budget_reallocation(packages, current_packages) if reallocation is not None: mutations.append(reallocation) reallocated_package_ids = set(reallocation.after) else: reallocated_package_ids = set() for index, package_patch in enumerate(packages): package_id = _package_id(package_patch) current_package = current_packages.get(package_id or "") mutations.extend( _decompose_package_patch( package_patch, index=index, package_id=package_id, current_package=current_package, skip_budget=package_id in reallocated_package_ids, ) ) for field_name in _TOP_LEVEL_UNMAPPED_MUTATION_FIELDS: if field_name in patch_dict: mutations.append( _unknown_mutation( (field_name,), after=patch_dict[field_name], raw=patch_dict[field_name], ) ) for field_name in _unknown_top_level_fields(patch_dict): mutations.append( _unknown_mutation( (field_name,), after=patch_dict[field_name], raw=patch_dict[field_name], ) ) return mutationsSplit an
update_media_buypatch into ordered logical mutations.patchmay be a generated PydanticUpdateMediaBuyRequestor a plain mapping.current_media_buyis optional; when supplied, the helper can promote coarse mutations into specific actions such asincrease_budget,decrease_budget,extend_flight,shorten_flight, andreallocate_budget. def disallowed_update_media_buy_mutations(patch: Any,
allowed_actions: Iterable[Any] | None,
current_media_buy: Any | None = None,
*,
allowed_modes: Iterable[str] | None = ('self_serve', 'conditional_self_serve')) ‑> list[UpdateMediaBuyMutation]-
Expand source code
def disallowed_update_media_buy_mutations( patch: Any, allowed_actions: Iterable[Any] | None, current_media_buy: Any | None = None, *, allowed_modes: Iterable[str] | None = SELF_SERVE_UPDATE_ACTION_MODES, ) -> list[UpdateMediaBuyMutation]: """Return action-gated mutations not covered by the supplied allowed actions. ``UNKNOWN_UPDATE_ACTION`` mutations stay visible in :func:`decompose_update_media_buy`, but are not treated as allowed-action failures because the protocol has no action mapping for them yet. """ return [ mutation for mutation in decompose_update_media_buy(patch, current_media_buy) if mutation.action != UNKNOWN_UPDATE_ACTION if not is_update_media_buy_mutation_allowed( mutation, allowed_actions, allowed_modes=allowed_modes, ) ]Return action-gated mutations not covered by the supplied allowed actions.
UNKNOWN_UPDATE_ACTIONmutations stay visible in :func:decompose_update_media_buy(), but are not treated as allowed-action failures because the protocol has no action mapping for them yet. def is_update_media_buy_mutation_allowed(mutation: UpdateMediaBuyMutation,
allowed_actions: Iterable[Any] | None,
*,
allowed_modes: Iterable[str] | None = ('self_serve', 'conditional_self_serve')) ‑> bool-
Expand source code
def is_update_media_buy_mutation_allowed( mutation: UpdateMediaBuyMutation, allowed_actions: Iterable[Any] | None, *, allowed_modes: Iterable[str] | None = SELF_SERVE_UPDATE_ACTION_MODES, ) -> bool: """Return whether ``allowed_actions`` contains a capability covering ``mutation``.""" return mutation.is_allowed_by(allowed_actions, allowed_modes=allowed_modes)Return whether
allowed_actionscontains a capability coveringmutation. def normalize_update_media_buy_allowed_actions(allowed_actions: Iterable[Any] | None,
*,
allowed_modes: Iterable[str] | None = None) ‑> tuple[str, ...]-
Expand source code
def normalize_update_media_buy_allowed_actions( allowed_actions: Iterable[Any] | None, *, allowed_modes: Iterable[str] | None = None, ) -> tuple[str, ...]: """Normalize action declarations to ordered action identifiers. Accepts any mix of action strings, generated enum values, wire dictionaries like ``{"action": "pause", "mode": "self_serve"}``, and generated ``MediaBuyAvailableAction`` models. When ``allowed_modes`` is provided, wire entries with a non-matching ``mode`` are filtered out. """ if allowed_actions is None: return () allowed_mode_set = set(allowed_modes) if allowed_modes is not None else None return _dedupe( action_name for action in allowed_actions if ( (action_name := _action_name(action)) is not None and _action_mode_matches(action, allowed_mode_set) ) )Normalize action declarations to ordered action identifiers.
Accepts any mix of action strings, generated enum values, wire dictionaries like
{"action": "pause", "mode": "self_serve"}, and generatedMediaBuyAvailableActionmodels. Whenallowed_modesis provided, wire entries with a non-matchingmodeare filtered out. def requested_update_media_buy_actions(patch: Any, current_media_buy: Any | None = None) ‑> tuple[str, ...]-
Expand source code
def requested_update_media_buy_actions( patch: Any, current_media_buy: Any | None = None, ) -> tuple[str, ...]: """Return ordered, de-duplicated actions requested by a patch.""" return _dedupe( mutation.action for mutation in decompose_update_media_buy(patch, current_media_buy) )Return ordered, de-duplicated actions requested by a patch.
Classes
class UpdateMediaBuyMutation (action: str,
field_paths: tuple[str, ...],
package_id: str | None = None,
before: Any = None,
after: Any = None,
raw: Any = None,
resolution: UpdateMutationResolution = 'fine',
allowed_action_candidates: tuple[str, ...] = <factory>)-
Expand source code
@dataclass(frozen=True, slots=True) class UpdateMediaBuyMutation: """A single logical mutation requested by an ``update_media_buy`` patch. ``action`` is the most specific action inferred from the patch. When the patch does not carry enough information to choose a fine-grained action, the helper emits the closest coarse action and marks ``resolution`` as ``"coarse"``. ``allowed_action_candidates`` are ordered from specific to broad. A caller can use them to match a mutation against either fine-grained capabilities such as ``increase_budget`` or coarse capabilities such as ``update_budget`` / ``update_packages``. """ action: str field_paths: tuple[str, ...] package_id: str | None = None before: Any = None after: Any = None raw: Any = None resolution: UpdateMutationResolution = "fine" allowed_action_candidates: tuple[str, ...] = field(default_factory=tuple) def is_allowed_by( self, allowed_actions: Iterable[Any] | None, *, allowed_modes: Iterable[str] | None = SELF_SERVE_UPDATE_ACTION_MODES, ) -> bool: """Return whether ``allowed_actions`` covers this mutation. ``allowed_actions`` may be a simple string list or the wire ``available_actions[]`` shape returned by ``get_media_buys``. By default, wire entries must be immediately executable (``self_serve`` or ``conditional_self_serve``); pass ``allowed_modes=None`` to ignore mode and check only action presence. """ allowed = set( normalize_update_media_buy_allowed_actions( allowed_actions, allowed_modes=allowed_modes, ) ) return any(candidate in allowed for candidate in self.allowed_action_candidates)A single logical mutation requested by an
update_media_buypatch.actionis the most specific action inferred from the patch. When the patch does not carry enough information to choose a fine-grained action, the helper emits the closest coarse action and marksresolutionas"coarse".allowed_action_candidatesare ordered from specific to broad. A caller can use them to match a mutation against either fine-grained capabilities such asincrease_budgetor coarse capabilities such asupdate_budget/update_packages.Instance variables
var action : str-
Expand source code
@dataclass(frozen=True, slots=True) class UpdateMediaBuyMutation: """A single logical mutation requested by an ``update_media_buy`` patch. ``action`` is the most specific action inferred from the patch. When the patch does not carry enough information to choose a fine-grained action, the helper emits the closest coarse action and marks ``resolution`` as ``"coarse"``. ``allowed_action_candidates`` are ordered from specific to broad. A caller can use them to match a mutation against either fine-grained capabilities such as ``increase_budget`` or coarse capabilities such as ``update_budget`` / ``update_packages``. """ action: str field_paths: tuple[str, ...] package_id: str | None = None before: Any = None after: Any = None raw: Any = None resolution: UpdateMutationResolution = "fine" allowed_action_candidates: tuple[str, ...] = field(default_factory=tuple) def is_allowed_by( self, allowed_actions: Iterable[Any] | None, *, allowed_modes: Iterable[str] | None = SELF_SERVE_UPDATE_ACTION_MODES, ) -> bool: """Return whether ``allowed_actions`` covers this mutation. ``allowed_actions`` may be a simple string list or the wire ``available_actions[]`` shape returned by ``get_media_buys``. By default, wire entries must be immediately executable (``self_serve`` or ``conditional_self_serve``); pass ``allowed_modes=None`` to ignore mode and check only action presence. """ allowed = set( normalize_update_media_buy_allowed_actions( allowed_actions, allowed_modes=allowed_modes, ) ) return any(candidate in allowed for candidate in self.allowed_action_candidates) var after : Any-
Expand source code
@dataclass(frozen=True, slots=True) class UpdateMediaBuyMutation: """A single logical mutation requested by an ``update_media_buy`` patch. ``action`` is the most specific action inferred from the patch. When the patch does not carry enough information to choose a fine-grained action, the helper emits the closest coarse action and marks ``resolution`` as ``"coarse"``. ``allowed_action_candidates`` are ordered from specific to broad. A caller can use them to match a mutation against either fine-grained capabilities such as ``increase_budget`` or coarse capabilities such as ``update_budget`` / ``update_packages``. """ action: str field_paths: tuple[str, ...] package_id: str | None = None before: Any = None after: Any = None raw: Any = None resolution: UpdateMutationResolution = "fine" allowed_action_candidates: tuple[str, ...] = field(default_factory=tuple) def is_allowed_by( self, allowed_actions: Iterable[Any] | None, *, allowed_modes: Iterable[str] | None = SELF_SERVE_UPDATE_ACTION_MODES, ) -> bool: """Return whether ``allowed_actions`` covers this mutation. ``allowed_actions`` may be a simple string list or the wire ``available_actions[]`` shape returned by ``get_media_buys``. By default, wire entries must be immediately executable (``self_serve`` or ``conditional_self_serve``); pass ``allowed_modes=None`` to ignore mode and check only action presence. """ allowed = set( normalize_update_media_buy_allowed_actions( allowed_actions, allowed_modes=allowed_modes, ) ) return any(candidate in allowed for candidate in self.allowed_action_candidates) var allowed_action_candidates : tuple[str, ...]-
Expand source code
@dataclass(frozen=True, slots=True) class UpdateMediaBuyMutation: """A single logical mutation requested by an ``update_media_buy`` patch. ``action`` is the most specific action inferred from the patch. When the patch does not carry enough information to choose a fine-grained action, the helper emits the closest coarse action and marks ``resolution`` as ``"coarse"``. ``allowed_action_candidates`` are ordered from specific to broad. A caller can use them to match a mutation against either fine-grained capabilities such as ``increase_budget`` or coarse capabilities such as ``update_budget`` / ``update_packages``. """ action: str field_paths: tuple[str, ...] package_id: str | None = None before: Any = None after: Any = None raw: Any = None resolution: UpdateMutationResolution = "fine" allowed_action_candidates: tuple[str, ...] = field(default_factory=tuple) def is_allowed_by( self, allowed_actions: Iterable[Any] | None, *, allowed_modes: Iterable[str] | None = SELF_SERVE_UPDATE_ACTION_MODES, ) -> bool: """Return whether ``allowed_actions`` covers this mutation. ``allowed_actions`` may be a simple string list or the wire ``available_actions[]`` shape returned by ``get_media_buys``. By default, wire entries must be immediately executable (``self_serve`` or ``conditional_self_serve``); pass ``allowed_modes=None`` to ignore mode and check only action presence. """ allowed = set( normalize_update_media_buy_allowed_actions( allowed_actions, allowed_modes=allowed_modes, ) ) return any(candidate in allowed for candidate in self.allowed_action_candidates) var before : Any-
Expand source code
@dataclass(frozen=True, slots=True) class UpdateMediaBuyMutation: """A single logical mutation requested by an ``update_media_buy`` patch. ``action`` is the most specific action inferred from the patch. When the patch does not carry enough information to choose a fine-grained action, the helper emits the closest coarse action and marks ``resolution`` as ``"coarse"``. ``allowed_action_candidates`` are ordered from specific to broad. A caller can use them to match a mutation against either fine-grained capabilities such as ``increase_budget`` or coarse capabilities such as ``update_budget`` / ``update_packages``. """ action: str field_paths: tuple[str, ...] package_id: str | None = None before: Any = None after: Any = None raw: Any = None resolution: UpdateMutationResolution = "fine" allowed_action_candidates: tuple[str, ...] = field(default_factory=tuple) def is_allowed_by( self, allowed_actions: Iterable[Any] | None, *, allowed_modes: Iterable[str] | None = SELF_SERVE_UPDATE_ACTION_MODES, ) -> bool: """Return whether ``allowed_actions`` covers this mutation. ``allowed_actions`` may be a simple string list or the wire ``available_actions[]`` shape returned by ``get_media_buys``. By default, wire entries must be immediately executable (``self_serve`` or ``conditional_self_serve``); pass ``allowed_modes=None`` to ignore mode and check only action presence. """ allowed = set( normalize_update_media_buy_allowed_actions( allowed_actions, allowed_modes=allowed_modes, ) ) return any(candidate in allowed for candidate in self.allowed_action_candidates) var field_paths : tuple[str, ...]-
Expand source code
@dataclass(frozen=True, slots=True) class UpdateMediaBuyMutation: """A single logical mutation requested by an ``update_media_buy`` patch. ``action`` is the most specific action inferred from the patch. When the patch does not carry enough information to choose a fine-grained action, the helper emits the closest coarse action and marks ``resolution`` as ``"coarse"``. ``allowed_action_candidates`` are ordered from specific to broad. A caller can use them to match a mutation against either fine-grained capabilities such as ``increase_budget`` or coarse capabilities such as ``update_budget`` / ``update_packages``. """ action: str field_paths: tuple[str, ...] package_id: str | None = None before: Any = None after: Any = None raw: Any = None resolution: UpdateMutationResolution = "fine" allowed_action_candidates: tuple[str, ...] = field(default_factory=tuple) def is_allowed_by( self, allowed_actions: Iterable[Any] | None, *, allowed_modes: Iterable[str] | None = SELF_SERVE_UPDATE_ACTION_MODES, ) -> bool: """Return whether ``allowed_actions`` covers this mutation. ``allowed_actions`` may be a simple string list or the wire ``available_actions[]`` shape returned by ``get_media_buys``. By default, wire entries must be immediately executable (``self_serve`` or ``conditional_self_serve``); pass ``allowed_modes=None`` to ignore mode and check only action presence. """ allowed = set( normalize_update_media_buy_allowed_actions( allowed_actions, allowed_modes=allowed_modes, ) ) return any(candidate in allowed for candidate in self.allowed_action_candidates) var package_id : str | None-
Expand source code
@dataclass(frozen=True, slots=True) class UpdateMediaBuyMutation: """A single logical mutation requested by an ``update_media_buy`` patch. ``action`` is the most specific action inferred from the patch. When the patch does not carry enough information to choose a fine-grained action, the helper emits the closest coarse action and marks ``resolution`` as ``"coarse"``. ``allowed_action_candidates`` are ordered from specific to broad. A caller can use them to match a mutation against either fine-grained capabilities such as ``increase_budget`` or coarse capabilities such as ``update_budget`` / ``update_packages``. """ action: str field_paths: tuple[str, ...] package_id: str | None = None before: Any = None after: Any = None raw: Any = None resolution: UpdateMutationResolution = "fine" allowed_action_candidates: tuple[str, ...] = field(default_factory=tuple) def is_allowed_by( self, allowed_actions: Iterable[Any] | None, *, allowed_modes: Iterable[str] | None = SELF_SERVE_UPDATE_ACTION_MODES, ) -> bool: """Return whether ``allowed_actions`` covers this mutation. ``allowed_actions`` may be a simple string list or the wire ``available_actions[]`` shape returned by ``get_media_buys``. By default, wire entries must be immediately executable (``self_serve`` or ``conditional_self_serve``); pass ``allowed_modes=None`` to ignore mode and check only action presence. """ allowed = set( normalize_update_media_buy_allowed_actions( allowed_actions, allowed_modes=allowed_modes, ) ) return any(candidate in allowed for candidate in self.allowed_action_candidates) var raw : Any-
Expand source code
@dataclass(frozen=True, slots=True) class UpdateMediaBuyMutation: """A single logical mutation requested by an ``update_media_buy`` patch. ``action`` is the most specific action inferred from the patch. When the patch does not carry enough information to choose a fine-grained action, the helper emits the closest coarse action and marks ``resolution`` as ``"coarse"``. ``allowed_action_candidates`` are ordered from specific to broad. A caller can use them to match a mutation against either fine-grained capabilities such as ``increase_budget`` or coarse capabilities such as ``update_budget`` / ``update_packages``. """ action: str field_paths: tuple[str, ...] package_id: str | None = None before: Any = None after: Any = None raw: Any = None resolution: UpdateMutationResolution = "fine" allowed_action_candidates: tuple[str, ...] = field(default_factory=tuple) def is_allowed_by( self, allowed_actions: Iterable[Any] | None, *, allowed_modes: Iterable[str] | None = SELF_SERVE_UPDATE_ACTION_MODES, ) -> bool: """Return whether ``allowed_actions`` covers this mutation. ``allowed_actions`` may be a simple string list or the wire ``available_actions[]`` shape returned by ``get_media_buys``. By default, wire entries must be immediately executable (``self_serve`` or ``conditional_self_serve``); pass ``allowed_modes=None`` to ignore mode and check only action presence. """ allowed = set( normalize_update_media_buy_allowed_actions( allowed_actions, allowed_modes=allowed_modes, ) ) return any(candidate in allowed for candidate in self.allowed_action_candidates) var resolution : Literal['fine', 'coarse', 'unknown']-
Expand source code
@dataclass(frozen=True, slots=True) class UpdateMediaBuyMutation: """A single logical mutation requested by an ``update_media_buy`` patch. ``action`` is the most specific action inferred from the patch. When the patch does not carry enough information to choose a fine-grained action, the helper emits the closest coarse action and marks ``resolution`` as ``"coarse"``. ``allowed_action_candidates`` are ordered from specific to broad. A caller can use them to match a mutation against either fine-grained capabilities such as ``increase_budget`` or coarse capabilities such as ``update_budget`` / ``update_packages``. """ action: str field_paths: tuple[str, ...] package_id: str | None = None before: Any = None after: Any = None raw: Any = None resolution: UpdateMutationResolution = "fine" allowed_action_candidates: tuple[str, ...] = field(default_factory=tuple) def is_allowed_by( self, allowed_actions: Iterable[Any] | None, *, allowed_modes: Iterable[str] | None = SELF_SERVE_UPDATE_ACTION_MODES, ) -> bool: """Return whether ``allowed_actions`` covers this mutation. ``allowed_actions`` may be a simple string list or the wire ``available_actions[]`` shape returned by ``get_media_buys``. By default, wire entries must be immediately executable (``self_serve`` or ``conditional_self_serve``); pass ``allowed_modes=None`` to ignore mode and check only action presence. """ allowed = set( normalize_update_media_buy_allowed_actions( allowed_actions, allowed_modes=allowed_modes, ) ) return any(candidate in allowed for candidate in self.allowed_action_candidates)
Methods
def is_allowed_by(self,
allowed_actions: Iterable[Any] | None,
*,
allowed_modes: Iterable[str] | None = ('self_serve', 'conditional_self_serve')) ‑> bool-
Expand source code
def is_allowed_by( self, allowed_actions: Iterable[Any] | None, *, allowed_modes: Iterable[str] | None = SELF_SERVE_UPDATE_ACTION_MODES, ) -> bool: """Return whether ``allowed_actions`` covers this mutation. ``allowed_actions`` may be a simple string list or the wire ``available_actions[]`` shape returned by ``get_media_buys``. By default, wire entries must be immediately executable (``self_serve`` or ``conditional_self_serve``); pass ``allowed_modes=None`` to ignore mode and check only action presence. """ allowed = set( normalize_update_media_buy_allowed_actions( allowed_actions, allowed_modes=allowed_modes, ) ) return any(candidate in allowed for candidate in self.allowed_action_candidates)Return whether
allowed_actionscovers this mutation.allowed_actionsmay be a simple string list or the wireavailable_actions[]shape returned byget_media_buys. By default, wire entries must be immediately executable (self_serveorconditional_self_serve); passallowed_modes=Noneto ignore mode and check only action presence.