EntitySession¶

EntitySession is a stateful wrapper around a StateMachine. It tracks the current state and context for one entity (one order, one user, one workflow instance) and provides a developer-friendly API on top of the raw FSM.

Creating a session¶

from pystator import StateMachine

machine = StateMachine.from_yaml("order_fsm.yaml")

# Create an instance for entity "order-123"
order = machine.create(context={"order_id": "order-123", "customer_id": "cust-456"})

machine.create() is a convenience wrapper — you can also instantiate directly:

from pystator import EntitySession

order = EntitySession(machine, context={"order_id": "order-123"})

Sending events¶

Synchronous¶

result = order.send("confirm")
print(result.success)       # True
print(result.source_state)  # "pending"
print(result.target_state)  # "confirmed"
print(result.trigger)       # "confirm"
print(result.error)         # None (or exception message on failure)

Pass payload data alongside the trigger:

result = order.send("ship", tracking_number="TRK-001", carrier="UPS")

The payload is merged into the context and available to guards and actions.

Auto-generated trigger methods¶

For each trigger defined in the YAML, EntitySession automatically creates a method on the instance:

order.confirm()              # same as order.send("confirm")
order.ship(tracking="TRK")   # same as order.send("ship", tracking="TRK")
order.cancel(reason="oos")   # same as order.send("cancel", reason="oos")

Trigger names in YAML must already be valid identifiers (lowercase a–z, digits, underscores). They are converted to Python method names by normalizing to snake_case (e.g. submit_order stays submit_order). Dots are not allowed in YAML trigger names.

Async¶

All async guards require asend:

result = await order.asend("confirm")

State queries¶

Auto-generated `is_<state>` properties¶

For each state in the YAML, EntitySession auto-generates a boolean property:

order.is_pending    # True  (state names become is_<snake_case>)
order.is_confirmed  # False
order.is_final      # False (not a terminal state)

Other properties¶

order.current_state      # "pending"
order.allowed_triggers   # ["confirm", "cancel"]
order.history            # list of TransitionResult from all sends
order.context            # current context dict (mutable)

Validate a target state (no trigger name)¶

When integration data only carries a status (state name), use evaluate_destination to see if that target is allowed from current_state with the session context. Optional context_overrides are merged into context for the check only (the session dict is not mutated).

check = order.evaluate_destination("confirmed")
check.allowed   # True if some transition can reach "confirmed" now

check = order.evaluate_destination("filled", context_overrides={"fill_qty": 100})
order.can_transition_to("filled")  # bool; same as check.allowed

If multiple triggers lead to the same destination, check.is_ambiguous (and check.metadata.get("ambiguous")) is True when more than one path succeeds—pick a policy in your app if you need a single trigger.

Snapshots — persist and restore¶

snapshot() serializes the entire session state into a plain dict you can store in a database, cache, or message queue. Restore it with from_snapshot():

# Persist
snap = order.snapshot()
# {
#   "current_state": "confirmed",
#   "context": {"order_id": "order-123", ...},
#   "machine_name": "order_lifecycle",
#   "machine_version": "1.0.0",
# }
redis_client.set("order:order-123", json.dumps(snap))

# Restore later
data = json.loads(redis_client.get("order:order-123"))
order = EntitySession.from_snapshot(machine, data)
print(order.current_state)  # "confirmed"

The snapshot is JSON-serializable (assuming your context values are serializable).

Deferred events¶

Set defer_unhandled=True when creating a session to queue events that don't match any transition in the current state, and replay them automatically after the next successful state change:

order = EntitySession(machine, context={}, defer_unhandled=True)

order.send("confirm")   # no transition for "confirm" in state "pending" → deferred
order.send("open")      # transitions to "open", then replays "confirm" → transitions to "confirmed"

When to use it: Useful for event-sourced systems where events may arrive out of order (e.g., a "payment_received" event arrives before the "order_placed" transition has been applied).

Inspecting deferred events:

print(order.deferred_events)  # list of (trigger, payload) tuples

order.clear_deferred()        # discard all queued events

`EventRecorder` — record and replay¶

EventRecorder wraps an EntitySession and records every event sent through it. The recording can then be replayed on a fresh session (e.g., to replay a sequence in tests or for debugging):

order = machine.create()
recorder = order.record()          # wrap with recorder

recorder.send("confirm")
recorder.send("ship", tracking="TRK-001")
recorder.send("deliver")

print(recorder.events)
# [RecordedEvent(trigger='confirm', payload={}, timestamp=...), ...]

# Replay on a fresh instance
fresh = machine.create()
result = recorder.replay(fresh)    # applies all recorded events
print(fresh.current_state)         # "delivered"

Saving and loading recordings¶

import json

# Save
recording_data = [e.to_dict() for e in recorder.events]
Path("recordings/order-123.json").write_text(json.dumps(recording_data))

# Load and replay
from pystator.instance import RecordedEvent

raw = json.loads(Path("recordings/order-123.json").read_text())
events = [RecordedEvent.from_dict(e) for e in raw]

fresh = machine.create()
for ev in events:
    fresh.send(ev.trigger, **ev.payload)

Thread safety¶

EntitySession.send() is thread-safe — it uses an internal threading.Lock. Do not share a single EntitySession across threads and async code; use either send (threads) or asend (async) exclusively within a given instance.

Example: order lifecycle¶

from pystator import StateMachine

machine = StateMachine.from_yaml("order_fsm.yaml")
order = machine.create(context={"order_id": "ORD-1"})

print(order.current_state)     # "pending"
print(order.is_pending)        # True

order.confirm()
print(order.current_state)     # "confirmed"

result = order.ship(carrier="FedEx", tracking="1234")
print(result.success)          # True
print(order.current_state)     # "shipped"

snap = order.snapshot()
# ... persist snap ...

restored = machine.create_from_snapshot(snap)
print(restored.current_state)  # "shipped"
print(restored.is_final)       # False