pystator 0.0.3
A configuration-driven, stateless finite state machine library for Python
Requires Python: >=3.10
Keywords: state-machine, fsm, workflow, declarative, event-driven

# PyStator

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

[![Python 3.10+](https://img.shields.io/badge/python-3.10+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

---

## Quick start (2 minutes)

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

```bash
pip install pystator
```

**1. Save this as `order_fsm.yaml`:**

```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:**

```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](#-concepts) and [Documentation](#-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](docs/guides/deployment.md).
- **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](#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](docs/guides/schedulers.md).

---

## 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

```bash
pip install pystator
```

### With API and UI

```bash
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:

```bash
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"`):

```bash
pip install pystator[recipes]
```

### Development

```bash
pip install -e ".[dev]"
```

---

## Quick start (extended)

### From a YAML file

```python
from pystator import StateMachine

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

### From a dict

```python
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

```python
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.

```python
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](docs/guides/schedulers.md).

### 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](docs/guides/state-stores.md) and persist before running actions; the library does not persist for you.

### REST API

With `pip install pystator[api]`:

```bash
# 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)](docs/getting-started/quickstart.md)** — Step-by-step first FSM and first API call
- **[Concepts](docs/guides/concepts.md)** — States, transitions, guards, actions, hierarchical and parallel
- **[Architecture](docs/guides/architecture.md)** — Design goals, core flow, sandwich pattern, components
- **[Configuration](docs/guides/configuration.md)** — Config file, environment, database (for API)
- **[Tutorials](docs/tutorials/index.md)** — Order workflow, API & UI, delayed transitions
- **[Examples](docs/examples.md)** — List of runnable examples with descriptions
- **[FSM config reference](docs/reference/fsm-config.md)** — Full YAML/JSON schema (meta, states, transitions, validation)
- **[API reference](docs/api.md)** — 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](examples/README.md) for how to run each. Tutorials in [docs/tutorials/](docs/tutorials/) walk through building an order workflow and using the API and UI.

---

## API reference (condensed)

### StateMachine

```python
# 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

```python
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

```python
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

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

---

## License

MIT — see [LICENSE](LICENSE).

---

## Links

- **Repository**: [GitHub](https://github.com/statfyi/pystator)
- **Issues**: [GitHub Issues](https://github.com/statfyi/pystator/issues)
- **Documentation**: [docs/](docs/) — quick start, guides, tutorials, examples