EntitySession API

Stateful per-entity FSM wrapper. Tracks current state, context, and history for a single entity.

See EntitySession guide for the full guide.

Classes

EntitySession

EntitySession(machine: StateMachine, *, entity_id: str | None = None, context: dict[str, Any] | None = None, initial_state: str | None = None, defer_unhandled: bool = False)

Stateful wrapper around a :class:StateMachine.

Tracks the current state and context for a single entity (e.g., one order, one user session). Provides:

• send(trigger, **payload) / asend(trigger, **payload)
• Auto-generated trigger methods (e.g., order.confirm())
• Auto-generated is_<state> properties (e.g., order.is_pending)
• current_state, context, is_final, workflow_status, allowed_triggers
• snapshot() for serializing the instance state (includes workflow_status).

Parameters:

Name	Type	Description	Default
machine	StateMachine	The StateMachine definition to use.	required
entity_id	str | None	Optional unique identifier (auto-generated UUID if omitted).	None
context	dict[str, Any] | None	Optional initial context dict.	None
initial_state	str | None	Override the initial state (defaults to machine's initial).	None
defer_unhandled	bool	If True, queue unmatched events for later replay.	False

Source code in src/pystator/instance.py

def __init__(
    self,
    machine: StateMachine,
    *,
    entity_id: str | None = None,
    context: dict[str, Any] | None = None,
    initial_state: str | None = None,
    defer_unhandled: bool = False,
) -> None:
    """Initialize a stateful entity session.

    Args:
        machine: The StateMachine definition to use.
        entity_id: Optional unique identifier (auto-generated UUID if omitted).
        context: Optional initial context dict.
        initial_state: Override the initial state (defaults to machine's initial).
        defer_unhandled: If True, queue unmatched events for later replay.
    """
    if entity_id is not None:
        from pystator.shared.entity_id_validator import validate_entity_id

        self._entity_id = validate_entity_id(entity_id)
    else:
        self._entity_id = str(uuid.uuid4())
    self._machine = machine
    self._context = dict(context or {})
    self._current_state = initial_state or machine.get_initial_state().name
    self._history: list[TransitionResult] = []
    self._lock = threading.Lock()
    self._async_lock: asyncio.Lock | None = None
    self._defer_unhandled = defer_unhandled
    self._deferred_events: list[tuple[str, dict[str, Any]]] = []

    # Dynamically generate trigger methods and is_* properties
    self._bind_trigger_methods()
    self._bind_state_properties()

entity_id property

entity_id: str

Unique identifier for this entity session.

current_state property

current_state: str

Current state name.

context property writable

context: dict[str, Any]

Current context (mutable reference).

is_final property

is_final: bool

True if current state is terminal.

workflow_status property

workflow_status: str

FSM workflow outcome: active, complete, or failed.

Derived from the current state's type (TERMINAL → complete, ERROR → failed, else active). Same semantics as persisted workflow_status on entity store rows.

allowed_triggers property

allowed_triggers: list[str]

Trigger names available from the current state.

allowed_transitions property

allowed_transitions: list[Any]

Transition definitions available from the current state.

history property

history: list[TransitionResult]

List of all transition results applied to this instance.

deferred_events property

deferred_events: list[tuple[str, dict[str, Any]]]

Pending deferred events (trigger, payload) not yet applied.

send

send(trigger: str, **payload: Any) -> TransitionResult

Send an event to the machine and apply the transition.

Thread-safe: uses an internal lock to serialize access.

When defer_unhandled=True was set, events that don't match any transition are queued and replayed after the next successful state change.

Parameters:

Name	Type	Description	Default
trigger	str	Event trigger name.	required
**payload	Any	Event payload data (merged into context for guards/actions).	{}

Returns:

Type	Description
TransitionResult	TransitionResult describing the outcome.

Source code in src/pystator/instance.py

def send(self, trigger: str, **payload: Any) -> TransitionResult:
    """Send an event to the machine and apply the transition.

    Thread-safe: uses an internal lock to serialize access.

    When ``defer_unhandled=True`` was set, events that don't match any
    transition are queued and replayed after the next successful state
    change.

    Args:
        trigger: Event trigger name.
        **payload: Event payload data (merged into context for guards/actions).

    Returns:
        TransitionResult describing the outcome.
    """
    with self._lock:
        return self._send_unlocked(trigger, payload)

asend async

asend(trigger: str, **payload: Any) -> TransitionResult

Async version of :meth:send (supports async guards).

Async-safe: uses an internal asyncio.Lock to serialize access.

Source code in src/pystator/instance.py

async def asend(self, trigger: str, **payload: Any) -> TransitionResult:
    """Async version of :meth:`send` (supports async guards).

    Async-safe: uses an internal asyncio.Lock to serialize access.
    """
    if self._async_lock is None:
        self._async_lock = asyncio.Lock()

    async with self._async_lock:
        return await self._asend_unlocked(trigger, payload)

apply_data

apply_data(**data: Any) -> list[TransitionResult]

Update context and evaluate when-route transitions.

Merges data into the session context, then runs the post-transition evaluation chain (auto-transitions + when-routes). No trigger is required -- the destination states' when clauses determine which transition (if any) fires.

Requires classification.auto_route: true in the machine meta.

Returns:

Type	Description
list[TransitionResult]	List of transition results that fired (may be empty).

Source code in src/pystator/instance.py

def apply_data(self, **data: Any) -> list[TransitionResult]:
    """Update context and evaluate when-route transitions.

    Merges *data* into the session context, then runs the post-transition
    evaluation chain (auto-transitions + when-routes). No trigger is
    required -- the destination states' ``when`` clauses determine which
    transition (if any) fires.

    Requires ``classification.auto_route: true`` in the machine meta.

    Returns:
        List of transition results that fired (may be empty).
    """
    with self._lock:
        return self._apply_data_unlocked(data)

aapply_data async

aapply_data(**data: Any) -> list[TransitionResult]

Async version of :meth:apply_data.

Source code in src/pystator/instance.py

async def aapply_data(self, **data: Any) -> list[TransitionResult]:
    """Async version of :meth:`apply_data`."""
    if self._async_lock is None:
        self._async_lock = asyncio.Lock()

    async with self._async_lock:
        self._context.update(data)
        ctx = dict(self._context)
        results = await self._machine._engine.async_evaluate_post_transition(
            self._current_state, ctx
        )
        for r in results:
            if r.success and r.target_state:
                self._machine._engine.apply_history(r)
                self._current_state = r.target_state
                self._history.append(r)
                await self._machine._afire_callbacks(r, ctx)
        return results

evaluate_destination

evaluate_destination(destination_state: str, context_overrides: dict[str, Any] | None = None) -> DestinationCheckResult

Check if destination_state is reachable from :attr:current_state.

Merges context_overrides into :attr:context for the check (session context is not mutated). Same semantics as :meth:StateMachine.evaluate_destination.

Source code in src/pystator/instance.py

def evaluate_destination(
    self,
    destination_state: str,
    context_overrides: dict[str, Any] | None = None,
) -> DestinationCheckResult:
    """Check if *destination_state* is reachable from :attr:`current_state`.

    Merges ``context_overrides`` into :attr:`context` for the check (session
    context is not mutated). Same semantics as
    :meth:`StateMachine.evaluate_destination`.
    """
    ctx = dict(self._context)
    if context_overrides:
        ctx.update(context_overrides)
    return self._machine.evaluate_destination(
        self._current_state,
        destination_state,
        ctx,
    )

can_transition_to

can_transition_to(destination_state: str, context_overrides: dict[str, Any] | None = None) -> bool

True if destination_state is reachable from the current state now.

Source code in src/pystator/instance.py

def can_transition_to(
    self,
    destination_state: str,
    context_overrides: dict[str, Any] | None = None,
) -> bool:
    """True if *destination_state* is reachable from the current state now."""
    return self.evaluate_destination(destination_state, context_overrides).allowed

clear_deferred

clear_deferred() -> None

Discard all deferred events.

Source code in src/pystator/instance.py

def clear_deferred(self) -> None:
    """Discard all deferred events."""
    self._deferred_events.clear()

snapshot

snapshot() -> dict[str, Any]

Serialize instance state for persistence.

Returns:

Type	Description
dict[str, Any]	Dict with current_state, context, and deferred events (serializable).

Source code in src/pystator/instance.py

def snapshot(self) -> dict[str, Any]:
    """Serialize instance state for persistence.

    Returns:
        Dict with current_state, context, and deferred events (serializable).
    """
    snap: dict[str, Any] = {
        "entity_id": self._entity_id,
        "current_state": self._current_state,
        "context": dict(self._context),
        "machine_name": self._machine.name,
        "machine_version": self._machine.version,
        "is_terminal": self._machine.is_terminal(self._current_state),
        "workflow_status": compute_workflow_status(
            self._machine, self._current_state
        ),
    }
    if self._deferred_events:
        snap["deferred_events"] = [
            {"trigger": t, "payload": p} for t, p in self._deferred_events
        ]
    return snap

from_snapshot classmethod

from_snapshot(machine: StateMachine, snapshot: dict[str, Any]) -> EntitySession

Restore an instance from a snapshot.

Parameters:

Name	Type	Description	Default
machine	StateMachine	The StateMachine definition.	required
snapshot	dict[str, Any]	Dict from :meth:snapshot.	required

Returns:

Type	Description
EntitySession	Restored EntitySession.

Source code in src/pystator/instance.py

@classmethod
def from_snapshot(
    cls,
    machine: StateMachine,
    snapshot: dict[str, Any],
) -> EntitySession:
    """Restore an instance from a snapshot.

    Args:
        machine: The StateMachine definition.
        snapshot: Dict from :meth:`snapshot`.

    Returns:
        Restored EntitySession.
    """
    instance = cls(
        machine=machine,
        entity_id=snapshot.get("entity_id"),
        context=snapshot.get("context"),
        initial_state=snapshot.get("current_state"),
        defer_unhandled=bool(snapshot.get("deferred_events")),
    )
    for ev in snapshot.get("deferred_events", []):
        instance._deferred_events.append((ev["trigger"], ev.get("payload", {})))
    return instance

record

record() -> EventRecorder

Return an EventRecorder that wraps this instance for replay support.

Source code in src/pystator/instance.py

def record(self) -> EventRecorder:
    """Return an EventRecorder that wraps this instance for replay support."""
    return EventRecorder(self)

EventRecorder

EventRecorder(instance: EntitySession)

Records events sent to an EntitySession for later replay.

Example::

order = machine.create()
recorder = order.record()
recorder.send("confirm", valid=True)
recorder.send("ship")

# Replay on a fresh instance
replayed = recorder.replay(machine)
assert replayed.current_state == order.current_state

Source code in src/pystator/instance.py

def __init__(self, instance: EntitySession) -> None:
    self._instance = instance
    self._events: list[RecordedEvent] = []
    self._initial_context: dict[str, Any] = dict(instance.context)
    self._initial_state: str = instance.current_state

events property

events: list[RecordedEvent]

All recorded events.

instance property

instance: EntitySession

The wrapped EntitySession.

send

send(trigger: str, **payload: Any) -> TransitionResult

Send an event (delegates to instance) and record it.

Source code in src/pystator/instance.py

def send(self, trigger: str, **payload: Any) -> TransitionResult:
    """Send an event (delegates to instance) and record it."""
    import time

    self._events.append(
        RecordedEvent(
            trigger=trigger,
            payload=dict(payload),
            timestamp=time.monotonic(),
        )
    )
    return self._instance.send(trigger, **payload)

replay

replay(machine: StateMachine) -> EntitySession

Create a fresh instance and replay all recorded events.

Source code in src/pystator/instance.py

def replay(self, machine: StateMachine) -> EntitySession:
    """Create a fresh instance and replay all recorded events."""
    new_instance = machine.create(
        context=dict(self._initial_context),
        initial_state=self._initial_state,
    )
    for event in self._events:
        new_instance.send(event.trigger, **event.payload)
    return new_instance

to_dict

to_dict() -> dict[str, Any]

Serialize the recording for storage.

Source code in src/pystator/instance.py

def to_dict(self) -> dict[str, Any]:
    """Serialize the recording for storage."""
    return {
        "initial_state": self._initial_state,
        "initial_context": self._initial_context,
        "events": [e.to_dict() for e in self._events],
    }

from_dict classmethod

from_dict(data: dict[str, Any], machine: StateMachine) -> EventRecorder

Deserialize a recording and attach to a fresh instance.

Source code in src/pystator/instance.py

@classmethod
def from_dict(cls, data: dict[str, Any], machine: StateMachine) -> EventRecorder:
    """Deserialize a recording and attach to a fresh instance."""
    instance = machine.create(
        context=data.get("initial_context", {}),
        initial_state=data.get("initial_state"),
    )
    recorder = cls(instance)
    recorder._events = [RecordedEvent.from_dict(e) for e in data.get("events", [])]
    return recorder

RecordedEvent dataclass

RecordedEvent(trigger: str, payload: dict[str, Any], timestamp: float)

A single recorded event with timestamp.

Import

# Recommended — via StateMachine factory
from pystator import StateMachine

machine = StateMachine.from_yaml("fsm.yaml")
session = machine.create(context={"id": "entity-1"})

# Direct import
from pystator import EntitySession
from pystator.instance import EntitySession, EventRecorder, RecordedEvent

Auto-generated methods and properties

EntitySession dynamically generates, for each state machine definition:

• is_<state_name> — boolean property; True when the current state matches
• <trigger_name>(**payload) — sends the named trigger (equivalent to .send(trigger, **payload))

YAML state and trigger names must already be valid lowercase identifiers (a–z, digits, underscores). EntitySession exposes them as: - is_<state> — e.g. state pending_new → is_pending_new - <trigger>(**payload) — e.g. trigger exchange_ack → exchange_ack()

See also

• EntitySession guide
• Orchestrator — multi-entity session management with persistence
• Stores