# pygenfsm
**Repository Path**: mirrors_serialx/pygenfsm
## Basic Information
- **Project Name**: pygenfsm
- **Description**: A minimal, clean, typed and asynchronous FSM implementation inspired by Erlang's gen_fsm
- **Primary Language**: Unknown
- **License**: MIT
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No
## Statistics
- **Stars**: 0
- **Forks**: 0
- **Created**: 2025-07-09
- **Last Updated**: 2026-07-18
## Categories & Tags
**Categories**: Uncategorized
**Tags**: None
## README
# ๐ pygenfsm
[](https://badge.fury.io/py/pygenfsm)
[](https://pypi.org/project/pygenfsm/)
[](https://opensource.org/licenses/MIT)
[](https://github.com/astral-sh/ruff)
[](https://github.com/microsoft/pyright)
[](https://github.com/serialx/pygenfsm)
**A minimal, clean, typed and async-native FSM (Finite State Machine) implementation for Python, inspired by Erlang's gen_fsm**
[Installation](#-installation) โข
[Quick Start](#-quick-start) โข
[Features](#-features) โข
[Examples](#-examples) โข
[API Reference](#-api-reference) โข
[Contributing](#-contributing)
---
## ๐ฏ Why pygenfsm?
Building robust state machines in Python often involves:
- ๐คฏ Complex if/elif chains that grow unmaintainable
- ๐ Implicit state that's hard to reason about
- ๐ Scattered transition logic across your codebase
- โ No type safety for states and events
- ๐ซ Mixing sync and async code awkwardly
**pygenfsm** solves these problems with a minimal, elegant API that leverages Python's type system and async capabilities.
## โจ Features
|
### ๐จ Clean API
```python
@fsm.on(State.IDLE, StartEvent)
def handle_start(fsm, event):
return State.RUNNING
```
|
### ๐ Async Native
```python
@fsm.on(State.RUNNING, DataEvent)
async def handle_data(fsm, event):
await process_data(event.data)
return State.DONE
```
|
|
### ๐ฏ Type Safe
```python
# Full typing with generics
FSM[StateEnum, EventType, ContextType]
```
|
### ๐ Zero Dependencies
```bash
# Minimal and fast
pip install pygenfsm
```
|
### Key Benefits
- **๐ Type-safe**: Full typing support with generics for states, events, and context
- **๐ญ Flexible**: Mix sync and async handlers in the same FSM
- **๐ฆ Minimal**: Zero dependencies, clean API surface
- **๐ Pythonic**: Decorator-based, intuitive design
- **๐ Async-native**: Built for modern async Python
- **๐ Context-aware**: Carry data between transitions
- **๐งฌ Cloneable**: Fork FSM instances for testing scenarios
- **๐๏ธ Builder pattern**: Late context injection support
## ๐ฆ Installation
```bash
# Using pip
pip install pygenfsm
# Using uv (recommended)
uv add pygenfsm
# Using poetry
poetry add pygenfsm
```
## ๐ Quick Start
### Basic Example
```python
import asyncio
from dataclasses import dataclass
from enum import Enum, auto
from pygenfsm import FSM
# 1. Define states as an enum
class State(Enum):
IDLE = auto()
RUNNING = auto()
DONE = auto()
# 2. Define events as dataclasses
@dataclass
class StartEvent:
task_id: str
@dataclass
class CompleteEvent:
result: str
# 3. Create FSM with initial state
fsm = FSM[State, StartEvent | CompleteEvent, None](
state=State.IDLE,
context=None, # No context needed for simple FSM
)
# 4. Define handlers with decorators
@fsm.on(State.IDLE, StartEvent)
def start_handler(fsm, event: StartEvent) -> State:
print(f"Starting task {event.task_id}")
return State.RUNNING
@fsm.on(State.RUNNING, CompleteEvent)
def complete_handler(fsm, event: CompleteEvent) -> State:
print(f"Task completed: {event.result}")
return State.DONE
# 5. Run the FSM
async def main():
await fsm.send(StartEvent(task_id="123"))
await fsm.send(CompleteEvent(result="Success!"))
print(f"Final state: {fsm.state}")
asyncio.run(main())
```
## ๐ฏ Core Concepts
### States, Events, and Context
pygenfsm is built on three core concepts:
| Concept | Purpose | Implementation |
|---------|---------|----------------|
| **States** | The finite set of states your system can be in | Python Enum |
| **Events** | Things that happen to trigger transitions | Dataclasses |
| **Context** | Data that persists across transitions | Any Python type |
### Handler Types
pygenfsm seamlessly supports both sync and async handlers:
```python
# Sync handler - for simple state transitions
@fsm.on(State.IDLE, SimpleEvent)
def sync_handler(fsm, event) -> State:
# Fast, synchronous logic
return State.NEXT
# Async handler - for I/O operations
@fsm.on(State.LOADING, DataEvent)
async def async_handler(fsm, event) -> State:
# Async I/O, network calls, etc.
data = await fetch_data(event.url)
fsm.context.data = data
return State.READY
```
## ๐ Examples
### Traffic Light System
```python
from enum import Enum, auto
from dataclasses import dataclass
from pygenfsm import FSM
class Color(Enum):
RED = auto()
YELLOW = auto()
GREEN = auto()
@dataclass
class TimerEvent:
"""Timer expired event"""
pass
@dataclass
class EmergencyEvent:
"""Emergency button pressed"""
pass
# Create FSM
traffic_light = FSM[Color, TimerEvent | EmergencyEvent, None](
state=Color.RED,
context=None,
)
@traffic_light.on(Color.RED, TimerEvent)
def red_to_green(fsm, event) -> Color:
print("๐ด โ ๐ข")
return Color.GREEN
@traffic_light.on(Color.GREEN, TimerEvent)
def green_to_yellow(fsm, event) -> Color:
print("๐ข โ ๐ก")
return Color.YELLOW
@traffic_light.on(Color.YELLOW, TimerEvent)
def yellow_to_red(fsm, event) -> Color:
print("๐ก โ ๐ด")
return Color.RED
# Emergency overrides from any state
for color in Color:
@traffic_light.on(color, EmergencyEvent)
def emergency(fsm, event) -> Color:
print("๐จ EMERGENCY โ RED")
return Color.RED
```
### Connection Manager with Retry Logic
```python
import asyncio
from dataclasses import dataclass, field
from enum import Enum, auto
from pygenfsm import FSM
class ConnState(Enum):
DISCONNECTED = auto()
CONNECTING = auto()
CONNECTED = auto()
ERROR = auto()
@dataclass
class ConnectEvent:
host: str
port: int
@dataclass
class ConnectionContext:
retries: int = 0
max_retries: int = 3
last_error: str = ""
fsm = FSM[ConnState, ConnectEvent, ConnectionContext](
state=ConnState.DISCONNECTED,
context=ConnectionContext(),
)
@fsm.on(ConnState.DISCONNECTED, ConnectEvent)
async def start_connection(fsm, event: ConnectEvent) -> ConnState:
print(f"๐ Connecting to {event.host}:{event.port}")
return ConnState.CONNECTING
@fsm.on(ConnState.CONNECTING, ConnectEvent)
async def attempt_connect(fsm, event: ConnectEvent) -> ConnState:
try:
# Simulate connection attempt
await asyncio.sleep(1)
if fsm.context.retries < 2: # Simulate failures
raise ConnectionError("Network timeout")
print("โ
Connected!")
fsm.context.retries = 0
return ConnState.CONNECTED
except ConnectionError as e:
fsm.context.retries += 1
fsm.context.last_error = str(e)
if fsm.context.retries >= fsm.context.max_retries:
print(f"โ Max retries reached: {e}")
return ConnState.ERROR
print(f"๐ Retry {fsm.context.retries}/{fsm.context.max_retries}")
return ConnState.CONNECTING
```
## ๐๏ธ Advanced Patterns
### Late Context Injection with FSMBuilder
Perfect for dependency injection and testing:
```python
from pygenfsm import FSMBuilder
# Define builder without context
builder = FSMBuilder[State, Event, AppContext](
initial_state=State.INIT
)
@builder.on(State.INIT, StartEvent)
async def initialize(fsm, event) -> State:
# Access context that will be injected later
await fsm.context.database.connect()
return State.READY
# Later, when dependencies are ready...
database = Database(connection_string)
logger = Logger(level="INFO")
# Build FSM with context
fsm = builder.build(AppContext(
database=database,
logger=logger,
))
```
### Cloning for Testing Scenarios
Test different paths without affecting the original:
```python
# Create base FSM
original_fsm = FSM[State, Event, Context](
state=State.INITIAL,
context=Context(data=[]),
)
# Clone for testing
test_scenario_1 = original_fsm.clone()
test_scenario_2 = original_fsm.clone()
# Run different scenarios
await test_scenario_1.send(SuccessEvent())
await test_scenario_2.send(FailureEvent())
# Original remains unchanged
assert original_fsm.state == State.INITIAL
```
## ๐ API Reference
### Core Classes
#### `FSM[S, E, C]`
The main FSM class with generic parameters:
- `S`: State enum type
- `E`: Event type (can be a Union)
- `C`: Context type
**Methods:**
- `on(state: S, event_type: type[E])`: Decorator to register handlers
- `async send(event: E) -> S`: Send event and transition state
- `send_sync(event: E) -> S`: Synchronous send (only for sync handlers)
- `clone() -> FSM[S, E, C]`: Create independent copy
- `replace_context(context: C) -> None`: Replace context
#### `FSMBuilder[S, E, C]`
Builder for late context injection:
- `on(state: S, event_type: type[E])`: Register handlers
- `build(context: C) -> FSM[S, E, C]`: Create FSM with context
### Best Practices
1. **Use sync handlers for:**
- Simple state transitions
- Pure computations
- Context updates
2. **Use async handlers for:**
- Network I/O
- Database operations
- File system access
- Long computations
3. **Event Design:**
- Make events immutable (use frozen dataclasses)
- Include all necessary data in events
- Use Union types for multiple events per state
4. **Context Design:**
- Keep context focused and minimal
- Use dataclasses for structure
- Avoid circular references
## ๐ค Contributing
We love contributions! Please see our [Contributing Guide](CONTRIBUTING.md) for details.
```bash
# Setup development environment
git clone https://github.com/serialx/pygenfsm
cd pygenfsm
uv sync
# Run tests
uv run pytest
# Run linting
uv run ruff check .
uv run pyright .
```
## ๐ Comparison with transitions
### Feature Comparison
| Feature | pygenfsm | transitions |
|---------|----------|-------------|
| **Event Data** | โ
First-class with dataclasses | โ Limited (callbacks, conditions) |
| **Async Support** | โ
Native async/await | โ No built-in support |
| **Type Safety** | โ
Full generics | โ ๏ธ Runtime checks only |
| **State Definition** | โ
Enums (type-safe) | โ ๏ธ Strings/objects |
| **Handler Registration** | โ
Decorators | โ Configuration dicts |
| **Context/Model** | โ
Explicit, typed | โ ๏ธ Implicit on model |
| **Dependencies** | โ
Zero | โ Multiple (six, etc.) |
| **Visualization** | โ Not built-in | โ
GraphViz support |
| **Hierarchical States** | โ No | โ
Yes (HSM) |
| **Parallel States** | โ No | โ
Yes |
| **State History** | โ No | โ
Yes |
| **Guards/Conditions** | โ ๏ธ In handler logic | โ
Built-in |
| **Callbacks** | โ ๏ธ In handlers | โ
before/after/prepare |
| **Size** | ~300 LOC | ~3000 LOC |
### When to Use Each
**Use pygenfsm when you need:**
- ๐ Strong type safety with IDE support
- ๐ Native async/await support
- ๐ฆ Zero dependencies
- ๐ฏ Event-driven architecture with rich data
- ๐ Modern Python patterns (3.11+)
- ๐งช Easy testing with full typing
**Use transitions when you need:**
- ๐ State diagram visualization
- ๐ Hierarchical states (HSM)
- โก Parallel state machines
- ๐ State history tracking
- ๐ Complex transition guards/conditions
- ๐๏ธ Legacy Python support
## ๐ Links
- **GitHub**: [github.com/serialx/pygenfsm](https://github.com/serialx/pygenfsm)
- **PyPI**: [pypi.org/project/pygenfsm](https://pypi.org/project/pygenfsm)
- **Documentation**: [Full API Docs](https://github.com/serialx/pygenfsm/wiki)
- **Issues**: [Report bugs or request features](https://github.com/serialx/pygenfsm/issues)
## ๐ License
MIT License - see [LICENSE](LICENSE) file for details.
---
Made with โค๏ธ by developers who love clean state machines