pystator 0.0.3

A configuration-driven, stateless finite state machine library for Python

PyStator

Configuration-driven, stateless finite state machines for Python: define behavior in YAML, compute transitions, run guards and actions.

---

Quick start (2 minutes)

Install, define a tiny state machine, and process one event. Copy-paste into a new terminal:

pip install pystator

1. Save this as 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

2. Run this Python:

from pystator import StateMachine

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

# Pure computation: current state + event → next state
result = machine.process("PENDING", "exchange_ack", {})

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

Then add guards (conditions) and actions (side effects), or use the REST API and UI with pip install pystator[api] (see Concepts and Documentation).

---

What is PyStator?

PyStator is a stateless finite state machine (FSM) library for Python. You define behavior in YAML or JSON; the engine computes transitions from (current state + event + context) and returns the next state and any actions to run. No internal state is held—ideal for APIs, workers, and distributed systems.

• Configuration-driven: Define states, transitions, guards, and actions in YAML/JSON with schema validation.
• Stateless: Pure computation—pass state in, get state and actions out; you persist state in your database.
• Hierarchical & parallel: Compound states, orthogonal regions, and statechart-style exit/enter semantics.
• Guards & actions: Conditional transitions (sync/async guards) and side effects executed after you persist the transition.
• Delayed transitions: Schedule transitions after a delay (asyncio, Redis, or Celery).
• Optional API & UI: REST API and web UI for validation, process, and machine CRUD (pip install pystator[api]).

---

Concepts

A short mental model so you know what to reach for.

Concept	What it is	When you use it
State	A node in the graph: initial, stable, terminal, or parallel.	Define the possible states of your entity (e.g. order: PENDING, OPEN, FILLED).
Transition	A rule: from state(s), on trigger event, to state; optional guards and actions.	Define how events move the entity between states.
Guard	A condition (sync or async) that must be true for the transition to fire.	Business rules (e.g. "full fill only if fill_qty >= order_qty").
Action	A side effect (sync or async) run after you persist the new state.	Notifications, DB updates, messaging—never for transition logic.
Context	A dict passed into process(current_state, trigger, context).	Event payload, entity data, and anything guards/actions need.

Flow from "just compute" to "full app":

  Option A: No persistence
  YAML FSM  →  StateMachine.from_yaml()  →  machine.process(state, event, context)
  You hold state in memory or pass it in each time.

  Option B: With persistence
  State store (DB/Redis)  →  load state  →  process()  →  Persist new state  →  Execute actions
  (Sandwich pattern: Load → Decide → Commit → Act). Use a [StateStore](docs/guides/state-stores.md) adapter.

  Option C: With API & UI
  pystator api  +  pystator ui serve  →  Validate configs, run process, manage machines via REST/UI

Start with Option A (Quick start above); add guards/actions when you need conditions and side effects; add API/UI when you want HTTP and a visual builder.

How PyStator runs (execution model)

PyStator is event-driven and stateless. You do not run a long-lived loop per FSM.

• Events drive transitions: When an event occurs (HTTP request, message from a queue, cron job, or a delayed-transition callback), you call process(current_state, trigger, context) or the Orchestrator’s async_process_event(entity_id, trigger, context) once. State is loaded from your store, the transition is computed, and you persist the new state and run actions.
• Scale by replicas: Run multiple copies of your API or worker (e.g. several pods). They share the same state store (database or Redis). Any replica can process any event for any entity—there is no “one pod per FSM.” See Deployment.
• Delayed transitions (after: "30s" in YAML) need a scheduler so that when the delay expires, something calls the orchestrator again. One process (or Redis/Celery) can track many delays for many entities and FSM types. See Choosing a scheduler below.

Glossary:

Term	Meaning
Machine (FSM definition)	The YAML/config: states and transitions. One per “type” (e.g. order_management).
Entity	One instance (e.g. order-123) with its own current state stored in the state store.
Replica	Another copy of your service (e.g. API pod). All replicas use the same state store and can process any entity.

Choosing a scheduler

Only needed if your FSM uses delayed transitions (after: "5s" or after: 5000 in a transition).

Need	Scheduler	Extra infra
Delayed transitions, single process or dev	AsyncioScheduler	None
Multiple replicas, HA, or survive restarts	RedisScheduler or CeleryScheduler	Redis or Celery (+ broker)

With AsyncioScheduler, one pod can track delays for all entities and all FSM types; delays are lost if that process exits. With Redis or Celery, delays are stored externally so any healthy worker can fire them. See Schedulers and delayed transitions.

---

Features

• Configuration-driven: YAML/JSON definitions with schema validation
• Stateless: Pure computation—no internal state
• Hierarchical states: Compound states, parent/child, LCA exit/enter
• Parallel states: Orthogonal regions—multiple active sub-states
• Delayed transitions: after: 5s or after: 5000 with pluggable schedulers (asyncio, Redis, Celery)
• Inline guards: expr: "fill_qty >= order_qty" in YAML (no Python for simple rules)
• Guards & actions: Sync and async; decorator-based registration
• Action parameters: Pass config from YAML into actions via params
• Timeouts: State-level timeout to a destination state
• Type-safe: Full type hints and PEP 561
• Retry & idempotency: Configurable retry, pluggable idempotency backends
• REST API & UI: Optional server and web UI for FSM validation and process

---

Installation

Core library

pip install pystator

With API and UI

pip install pystator[api]

Installs FastAPI, Uvicorn, and PyJWT for the REST API (and optional auth). The UI is served by the same server when you run pystator ui serve (requires a built UI; see below).

With UI (development)

To build and serve the Next.js UI from source:

pip install pystator[api,ui]
cd src/pystator/ui && npm install && npm run build
pystator ui serve   # Serves UI + proxies API

From the project root you can also run pystator ui dev for hot-reload development.

Optional: recipes (inline guards)

For inline guard expressions in YAML (expr: "qty > 0"):

pip install pystator[recipes]

Development

pip install -e ".[dev]"

---

Quick start (extended)

From a YAML file

from pystator import StateMachine

machine = StateMachine.from_yaml("order_fsm.yaml")
result = machine.process("PENDING", "exchange_ack", {})

From a dict

from pystator import StateMachine

config = {
    "meta": {"version": "1.0.0", "machine_name": "my_fsm", "strict_mode": True},
    "states": [
        {"name": "A", "type": "initial"},
        {"name": "B", "type": "stable"},
        {"name": "C", "type": "terminal"},
    ],
    "transitions": [
        {"trigger": "go", "source": "A", "dest": "B"},
        {"trigger": "done", "source": "B", "dest": "C"},
    ],
}
machine = StateMachine.from_dict(config)
result = machine.process("A", "go", {})

With guards and actions

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", "execution_report", {"fill_qty": 100, "order_qty": 100})
if result.success:
    # 1. Persist state change to your DB
    # 2. Then run actions
    executor.execute(result, {"fill_qty": 100, "order_qty": 100})

Orchestrator and delayed transitions

For persistence plus delayed transitions (after: "5s" in YAML), use the Orchestrator with a state store and a scheduler. The orchestrator runs the full loop: load state → process → persist → schedule delayed transitions → execute actions.

import asyncio
from pystator import StateMachine, Orchestrator, GuardRegistry, ActionRegistry
from pystator.state_stores import InMemoryStateStore
from pystator.scheduler import AsyncioScheduler

machine = StateMachine.from_yaml("my_fsm.yaml")  # has a transition with after: "2s"
store = InMemoryStateStore()
guards = GuardRegistry()
actions = ActionRegistry()
scheduler = AsyncioScheduler()

orchestrator = Orchestrator(
    machine=machine, state_store=store, guards=guards, actions=actions, scheduler=scheduler
)

async def main():
    await orchestrator.async_process_event("entity-1", "start", {})
    await asyncio.sleep(2.5)  # delayed transition fires
    await orchestrator.close()

asyncio.run(main())

No extra infrastructure: AsyncioScheduler keeps delays in memory. For multiple replicas or restarts, use RedisScheduler or CeleryScheduler.

Common pitfalls

• Guards vs actions: Use guards for pure logic (can this transition run?). Use actions for side effects (notify, persist to another system). Don’t put side effects in guards.
• AsyncioScheduler: Delays are in-memory; they are lost if the process exits. Use Redis or Celery for production or multiple replicas.
• State store: With Option B, you must implement a StateStore and persist before running actions; the library does not persist for you.

REST API

With pip install pystator[api]:

# Start API (default: http://localhost:8000)
pystator api
# or: uvicorn pystator.api.main:app --reload
# Optional: use pystator.cfg for database and auth (copy pystator.cfg.example to pystator.cfg)

Endpoint	Method	Description
/health	GET	Health check
/api/v1/auth/me	GET	Current user (auth)
/api/v1/validate	POST	Validate FSM config
/api/v1/process	POST	Compute transition
/api/v1/machines	GET/POST	List/create machines
/api/v1/machines/{id}	GET/PUT/DELETE	CRUD machine

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

---

Documentation

• Quick start (detailed) — Step-by-step first FSM and first API call
• Concepts — States, transitions, guards, actions, hierarchical and parallel
• Architecture — Design goals, core flow, sandwich pattern, components
• Configuration — Config file, environment, database (for API)
• Tutorials — Order workflow, API & UI, delayed transitions
• Examples — List of runnable examples with descriptions
• FSM config reference — Full YAML/JSON schema (meta, states, transitions, validation)
• API reference — StateMachine, Orchestrator, schedulers, execution modes

---

Examples and tutorials

Runnable examples live in the examples/ directory:

Example	Description
basic_usage.py + order_fsm.yaml	Order lifecycle: load FSM, register guards/actions, process events
day_trading_example.py + day_trading_fsm.yaml	Parallel states (trading + risk monitor + data feed)
portfolio_optimization_example.py + portfolio_optimization_fsm.yaml	Hierarchical states and workflows

See examples/README.md for how to run each. Tutorials in docs/tutorials/ walk through building an order workflow and using the API and UI.

---

API reference (condensed)

StateMachine

# Create
machine = StateMachine.from_yaml("config.yaml")
machine = StateMachine.from_dict(config_dict)

# Process (sync)
result = machine.process(current_state, trigger, context)

# Process (async, for async guards)
result = await machine.async_process(current_state, trigger, context)

# Parallel states
config = machine.enter_parallel_state("parallel_state_name")
config, results = machine.process_parallel(config, event, context)

# Queries
machine.get_initial_state()
machine.get_available_transitions("STATE_NAME")

TransitionResult

result.success          # bool
result.source_state     # str
result.target_state     # str | None
result.trigger          # str
result.all_actions      # tuple[str, ...]  (exit + transition + enter)
result.error            # FSMError | None

Guards and actions

guards = GuardRegistry()
guards.register("name", lambda ctx: bool)
@guards.decorator("name")
def my_guard(ctx: dict) -> bool: ...

actions = ActionRegistry()
actions.register("name", lambda ctx: None)
@actions.decorator()
def my_action(ctx: dict) -> None: ...

machine.bind_guards(guards)
executor = ActionExecutor(actions)
executor.execute(transition_result, context)
# Async: await executor.async_execute_parallel(result, context)

---

Development

pip install -e ".[dev]"
pytest
mypy src/
ruff check . && ruff format .

---

License

MIT — see LICENSE.

---

Links

• Repository: GitHub
• Issues: GitHub Issues
• Documentation: docs/ — quick start, guides, tutorials, examples