Skip to content

Quick Start

Get from zero to a working state machine in a few minutes.

1. Install

pip install pystator

2. Define a state machine (YAML)

Create a file order_fsm.yaml:

meta:
  version: "1.0.0"
  machine_name: "order_management"
  strict_mode: true

states:
  - name: PENDING
    type: initial
  - name: OPEN
    type: stable
  - name: FILLED
    type: terminal

transitions:
  - trigger: exchange_ack
    source: PENDING
    dest: OPEN
  - trigger: fill
    source: OPEN
    dest: FILLED
  • States: initial (entry point), stable (normal), terminal (end).
  • Transitions: On trigger from source state(s) to dest state.

3. Load and process in Python

from pystator import StateMachine

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

# Process one event (pure computation)
result = machine.process("PENDING", "exchange_ack", {})

print(result.success)        # True
print(result.source_state)   # PENDING
print(result.target_state)   # OPEN

process(current_state, trigger, context) is pure: no side effects, no persistence. You pass in the current state (e.g. from your database); you get back the next state and any actions to run.

For per-entity state management in memory, EntitySession is the idiomatic approach. It holds current state and context for one entity (one order, one trade, one user):

from pystator import StateMachine, EntitySession

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

# Create a stateful instance for one entity
order = machine.create(context={"order_id": "order-123", "order_qty": 100})

print(order.current_state)   # PENDING
print(order.is_pending)      # True  — auto-generated property for every state
print(order.is_final)        # False

# Send events; payload merges into context
order.send("exchange_ack")
print(order.current_state)   # OPEN

order.send("fill", fill_qty=100)
print(order.current_state)   # FILLED
print(order.is_final)        # True

# Auto-generated trigger methods (equivalent to send):
# order.exchange_ack()
# order.fill(fill_qty=100)

# Serialize/restore state (e.g. to/from Redis or a DB)
snapshot = order.snapshot()
restored = EntitySession.from_snapshot(machine, snapshot)
print(restored.current_state)  # FILLED

EntitySession auto-generates instance.is_<state> properties and instance.<trigger>() methods for every state and trigger defined in the YAML.

5. Add guards (optional)

Guards are conditions that must be true for a transition to fire. Register them in Python and bind to the machine:

from pystator import StateMachine, GuardRegistry

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

guards = GuardRegistry()
guards.register("is_full_fill", lambda ctx: ctx.get("fill_qty", 0) >= ctx.get("order_qty", 1))
machine.bind_guards(guards)

# Only succeeds if context satisfies the guard
result = machine.process("OPEN", "fill", {"fill_qty": 100, "order_qty": 100})

You can also define inline guards in YAML with expr: "fill_qty >= order_qty" (requires pystator[recipes]).

6. Add actions (optional)

Actions are side effects (notifications, DB updates) run after you persist the new state. Register them and run via ActionExecutor:

from pystator import StateMachine, GuardRegistry, ActionRegistry
from pystator.actions import ActionExecutor

machine = StateMachine.from_yaml("order_fsm.yaml")
guards = GuardRegistry()
guards.register("is_full_fill", lambda ctx: ctx.get("fill_qty", 0) >= ctx.get("order_qty", 1))
machine.bind_guards(guards)

actions = ActionRegistry()
actions.register("update_positions", lambda ctx: print("Positions updated"))
executor = ActionExecutor(actions)

result = machine.process("OPEN", "fill", {"fill_qty": 100, "order_qty": 100})
if result.success:
    # 1. Persist state change (your DB)
    # 2. Then execute actions
    executor.execute(result, {"fill_qty": 100, "order_qty": 100})

For a full persistence setup (load/save state by entity id, delayed transitions), use a StateStore with the Orchestrator; see State stores and Schedulers.

7. Hooks and metrics (optional)

Attach lifecycle hooks to collect metrics or add structured logging:

from pystator import StateMachine, LoggingHook, MetricsCollector, TransitionObserver

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

observer = TransitionObserver()
observer.add_hook(LoggingHook())   # structured logging

metrics = MetricsCollector()
observer.add_hook(metrics)
# Use observer at your process/orchestrator level when processing events

machine.process("PENDING", "exchange_ack", {})
machine.process("OPEN", "fill", {"fill_qty": 100, "order_qty": 100})

summary = metrics.get_summary()
print(summary["success_rate"])   # 1.0
print(summary["duration"])       # {"avg_ms": ..., "p95_ms": ...}

8. Error handling

Catch specific exception subclasses, or FSMError for everything:

from pystator import (
    FSMError, GuardRejectedError, InvalidTransitionError,
    UndefinedTriggerError, TerminalStateError, ErrorCode
)

try:
    result = machine.process(current_state, trigger, context)
except GuardRejectedError as e:
    print(f"Guard '{e.guard_name}' blocked in state '{e.current_state}'")
except TerminalStateError:
    print("Entity already in a terminal state")
except FSMError as e:
    print(f"Error code: {e.code}")  # ErrorCode enum member

# Or use the non-exception path:
result = machine.process(current_state, trigger, context)
if not result.success:
    print(result.error)     # FSMError subclass
    print(result.metadata)  # {"reason": "guard_rejected", "guard_name": "is_full_fill", ...}

9. Use the REST API (optional)

With pip install pystator[api]:

pystator api
# or: uvicorn pystator.api.main:app --reload
  • Validate config: POST /api/v1/validate with {"config": {...}}
  • Process event: POST /api/v1/process with {"config": {...}, "current_state": "OPEN", "trigger": "fill", "context": {...}}
  • Machines: GET/POST /api/v1/machines to list or create stored machines (requires DB)

API docs: http://localhost:8000/docs

Next steps