GitHub - serialx/pygenfsm: A minimal, clean, typed and asynchronous FSM implementation inspired by Erlang's gen_fsm

A minimal, clean, typed and async-native FSM (Finite State Machine) implementation for Python, inspired by Erlang's gen_fsm

Installation • Quick Start • Features • Examples • API Reference • Contributing

# Using pip
pip install pygenfsm

# Using uv (recommended)
uv add pygenfsm

# Using poetry
poetry add pygenfsm

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

# 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

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

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

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

# 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

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