pystator 0.0.2

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
  Load state from DB  →  process()  →  Persist new state  →  Execute actions
  (Sandwich pattern: Load → Decide → Commit → Act)

  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.

---

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})

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
• Configuration — Config file, environment, database (for API)
• Tutorials — Order workflow, API & UI, delayed transitions
• Examples — List of runnable examples with descriptions
• 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