angee.base.transitions

Guarded transition methods for StateField columns.

API contract:

StateTransitions(field, graph, policy_setting=None) opts one StateField into guarding and binds it to the declared source-to-target graph. Graph keys are source values or source lists; graph values are target values or target lists. Values are normalized through the field, so callers may use enum members, stored values, or the enum member names the field accepts.

policy_setting names a composed Django settings key holding a policy overlay {"allow": [[source, target], ...], "deny": [[source, target], ...]} whose edge values are enum values (strings). The overlay is merged over the declared graph when a transition runs — allow adds edges, deny removes them, deny winning over allow — so a deployment enables or disables specific edges through composed settings (autoconfig defaults overridden by the project settings.yaml) with no code change. Reading the overlay at call time is what lets composed settings and test overrides take effect. A per-scope resolution seam (resolving the overlay through the instance's owning scope) is reserved but not built.

@transition(field, source=..., target=..., conditions=[...], on_success=..., policy=...) decorates the model's own transition methods. source is a single value or a list of values. Conditions are pure condition(instance) callables; a false condition raises TransitionNotAllowed and the method body does not run. on_success is an explicit hook(instance, source, target) callback; there is no signal dispatch. policy marks a transition whose edge is governed by the declaration's policy_setting: such an edge need not appear in the declared graph (it may be default-disabled), so class construction does not require it, and calling the method while the policy leaves the edge disabled raises TransitionNotAllowed. A policy-enabled edge still runs the declared guards and conditions. There is no transition-driven verb/surface registry today (transitions are surfaced by hand-authored mutations), so the marker carries the edge's policy name for a future surface to exclude a disabled verb; the guard behavior is what ships now.

The decorated method body runs after the source/graph/condition checks and before the target write. The primitive owns the state write and then calls on_success. It does not save the model; transition methods remain ordinary model methods and own any persistence of non-state fields. Illegal transitions raise TransitionNotAllowed with the field, source, and target in the message.

Direct Python assignment to a guarded field is rejected at descriptor level after initial model construction. The descriptor still permits initial loading, idempotent field normalization, and the primitive's own target write, so existing StateField users are untouched unless they declare StateTransitions.

TransitionNotAllowed

class TransitionNotAllowed(Exception)

Raised when a guarded transition or direct guarded-field write is illegal.

TransitionActionSpec

@dataclass(frozen=True, slots=True)
class TransitionActionSpec()

Public projection of one declared transition method.

transition

def transition(
    field: StateField,
    *,
    source: Any,
    target: Any,
    conditions: list[Condition] | tuple[Condition, ...] | None = None,
    on_success: SuccessHook | None = None,
    policy: str | None = None
) -> Callable[[TransitionMethod], TransitionMethod]

Decorate a model method as a guarded transition for field.

The matching StateTransitions declaration validates the source and target against its declared graph when the model class is built. At call time the wrapper checks the current source against the policy-merged graph, evaluates pure conditions, runs the method body, writes the target state, and invokes the explicit success hook. The save_state is the common on_success hook for ordinary models that persist the transitioned state plus fields touched by the method body. policy marks a policy-governed edge (see the module docstring): it is validated against the declaration's policy_setting overlay at call time rather than required in the declared graph at class-build time.

StateTransitions

class StateTransitions()

Declaration that guards one StateField and its model methods.

Declare this in the model body after the field it guards:

status_transitions = StateTransitions(status, {Status.DRAFT: [Status.READY]})

Then decorate the model's own methods with @transition(status, ...). The declaration installs the guarded descriptor only for that opted-in field and validates decorated methods against the declared source-to-target graph. An optional policy_setting names a composed settings key whose overlay enables or disables edges over that graph at call time (see the module docstring). It is intentionally local to the model class: no global registry, no off-model flow object, and no hidden success dispatch.

__init__

def __init__(field: StateField,
             graph: Mapping[Any, Any],
             policy_setting: str | None = None) -> None

Store the field, declared graph, and optional policy-overlay setting.

contribute_to_class

def contribute_to_class(cls: type[models.Model], name: str) -> None

Attach the declaration, descriptor guard, method metadata, and helper.

action_specs

@classmethod
def action_specs(
        cls, model: type[models.Model]) -> tuple[TransitionActionSpec, ...]

Return frozen action specs declared on model, sorted by method name.

revalidate_for

def revalidate_for(cls: type[models.Model]) -> None

Re-run this declaration's class-build validation against cls's MRO.

contribute_to_class validates only the declaring class's own methods and installs the guarded descriptor. The composer calls this instead after it reorders a materialized child's bases (child_overrides_parent), so the flipped MRO is proven to still satisfy the same class-build checks — every reachable @transition method still guards a declared or policy edge. It validates without mutating cls (no descriptor install).

Deliberately stricter than runtime dispatch: at call time a decorated method runs under its own bound declaration (spec.declaration), but here every reachable spec that matches this declaration's field is checked against this declaration's graph — including one bound to another declaration for the same field. This over-approximation is what gives the flip guard teeth: it rejects a reorder that brings a transition method and a narrower same-field declaration together, at build time, rather than trusting the emitted MRO to dispatch it to a graph that happens to allow it. With the usual single declaration per field the two are equivalent (a field's specs are all bound to the one declaration that guards it).

run

def run(instance: models.Model, spec: _TransitionSpec,
        method: TransitionMethod, args: tuple[Any, ...],
        kwargs: dict[str, Any]) -> Any

Execute one decorated transition method under this declaration.

force_state

def force_state(instance: models.Model, target: Any, *, reason: str) -> None

Force this field to target while bypassing the declared graph.

This is the explicit escape hatch for state targets that are data-dependent rather than graph-derived: recovery code may need to reconcile a lifecycle from persisted external resource names, and unsaved instances may need an initial target before the row exists. It still uses the same guarded field write and, for saved rows, the same save_state concurrency check as a declared transition. Callers must pass a concrete reason so every graph bypass is greppable and intentional.

not_allowed

def not_allowed(source: Any, target: Any) -> NoReturn

Raise the primitive's standard TransitionNotAllowed for this field.

_GuardedStateDescriptor

class _GuardedStateDescriptor(DeferredAttribute)

Descriptor that blocks direct changes to an opted-in state field.

__set__

def __set__(instance: models.Model, value: Any) -> None

Store only initial, idempotent, or transition-owned values.

save_state

def save_state(instance: models.Model, source: Any, target: Any) -> None

Persist a transition-owned state change plus method-touched fields.

An optimistic concurrency guard brackets the write: the row is locked and its committed state re-read, and a divergence from source (a concurrent transition already advanced the row) raises :class:TransitionNotAllowed rather than silently double-applying the transition — the document-row race that would post a ledger entry twice. The guard and the save share one transaction, and the guard only verifies; instance.save still performs the source -> target column write, so post_save receivers, audit stamping, changes publishers, and pre-save change trackers observe the real old->new transition.

revalidate_transition_metadata

def revalidate_transition_metadata(cls: type[models.Model]) -> None

Re-validate every StateTransitions declaration reachable on cls.

The composer calls this for a materialized child whose base order it flipped (child_overrides_parent): each declaration was validated when the class that declared it was built, but the reordered MRO must still satisfy the same class-build checks. Raises ImproperlyConfigured (via the declaration) when the reorder leaves a transition method guarding an undeclared edge.