fsm-cub

Most Python state machine libraries try to be magical: complex DSLs, hierarchical states, conditional guards, side effect frameworks, async whatnot.

fsm-cub is the opposite, it's a small, opinionated, faithfully Pythonic implementation of a simple FSM.

Define states as a StrEnum or IntEnum, declare allowed transitions, attach optional on_enter/on_exit callbacks, and you're done.

Optional event bound subclass publishes typed StateChanged events for cross system reactions. Pydantic-backed so it round trips through seralization cleanly. ~250 lines of code you can read in five minutes.

Installation

pip install fsm-cub

With uv:

uv pip install fsm-cub

Usage

Choosing a state type

Be explicit about your enum's base type. The state parameter S should be StrEnum (for string-valued states) or IntEnum (for integer-valued states) — both round-trip cleanly through model_dump_json() / model_validate_json().

A plain Enum works in memory (construction, transitions, callbacks, event emission all behave normally), but the JSON save round-trip will fail with a ValidationError because JSON object keys are always strings and Pydantic can't recover non-string keys back into a plain Enum. The fix is one line: change class Color(Enum) to class Color(IntEnum) and you're done.

Rule of thumb: StrEnum if your states are conceptually labels ("locked", "open"), IntEnum if they're conceptually ordered or compact ids (1, 2, 3).

Basic — a standalone FSM

Three steps: define your states (here as a StrEnum), declare which transitions are allowed, then transition.

from enum import StrEnum

from fsm_cub import FSM


class DoorState(StrEnum):
    LOCKED = "locked"
    CLOSED = "closed"
    OPEN = "open"


door = FSM[DoorState](initial=DoorState.LOCKED)
door.allow_many(
    (DoorState.LOCKED, DoorState.CLOSED),
    (DoorState.CLOSED, DoorState.OPEN),
    (DoorState.OPEN, DoorState.CLOSED),
    (DoorState.CLOSED, DoorState.LOCKED),
)

door.transition(DoorState.CLOSED)  # True — moves LOCKED → CLOSED
door.transition(DoorState.OPEN)    # True — moves CLOSED → OPEN
door.transition(DoorState.LOCKED)  # False — OPEN → LOCKED isn't allowed

door.is_(DoorState.OPEN)           # True
door.is_any(DoorState.OPEN, DoorState.CLOSED)  # True
door.transitions_from(DoorState.OPEN)          # {DoorState.CLOSED}

By default, invalid transitions return False and silently fail — convenient for games-in-progress where you'd rather see the symptom than crash. Flip strict mode on in tests (or via FSM_CUB_STRICT=1) and invalid transitions raise ValueError instead:

from fsm_cub import set_strict

set_strict(True)
door.transition(DoorState.LOCKED)  # raises ValueError

With callbacks — on_enter and on_exit

Hooks fire in on_exit(old) → on_enter(new) order. They're behavior, not state — they don't serialize through model_dump_json(), so you re-register them after loading a save.

door.on_enter(DoorState.OPEN, lambda: print("the door swings open"))
door.on_exit(DoorState.LOCKED, lambda: print("unlocking..."))

door.transition(DoorState.CLOSED)  # prints "unlocking..."
door.transition(DoorState.OPEN)    # prints "the door swings open"

Quality-of-life helpers

allow_bidirectional(a, b) — sugar for the common (a, b) + (b, a) symmetric case:

door.allow_bidirectional(DoorState.CLOSED, DoorState.OPEN)
# equivalent to: door.allow_many((CLOSED, OPEN), (OPEN, CLOSED))

reset() — return to initial while firing exit/enter callbacks. Bypasses the transition table on purpose (reset is structural, not user-puzzle). Use this for game-over restarts or scene changes; use force() only when you also want to skip callbacks (e.g. loading a save).

door.transition(DoorState.CLOSED)
door.transition(DoorState.OPEN)
door.reset()  # fires on_exit(OPEN), on_enter(LOCKED); current is back to LOCKED

track_history(max_len=64) — opt-in bounded ring buffer of (from, to) tuples. Off by default (zero memory cost when unused). Great for debugging "wait, how did we get here?":

door.track_history(max_len=16)
door.transition(DoorState.CLOSED)
door.transition(DoorState.OPEN)
door.history  # (('locked', 'closed'), ('closed', 'open'))

fsm.history always reads safely — returns an empty tuple if tracking was never enabled, so you don't have to check first. clear_history() empties the buffer without disabling tracking.

Save round-trips — Pydantic out of the box

FSM is a Pydantic BaseModel, so its transition table, initial state, and current state serialize cleanly. Callbacks don't (they're PrivateAttr), so you re-register them on load.

blob = door.model_dump_json()
revived = FSM[DoorState].model_validate_json(blob)
assert revived.current == door.current
assert revived.transitions_from(DoorState.CLOSED) == door.transitions_from(DoorState.CLOSED)

Advanced — event-bound FSMs for cross-system reactions

When other systems need to react to a transition without the FSM knowing they exist, use EventBoundFSM. It publishes a typed StateChanged[S] payload to a Topic on every successful transition.

from enum import StrEnum

from fsm_cub import EventBoundFSM, StateChanged, Topic


class DoorState(StrEnum):
    LOCKED = "locked"
    CLOSED = "closed"
    OPEN = "open"


door_events: Topic[StateChanged[DoorState]] = Topic("doors")


@door_events.subscribe
def log_change(ev: StateChanged[DoorState]) -> None:
    print(f"door #{ev.id}: {ev.from_state} -> {ev.to_state}")


door = EventBoundFSM[DoorState](initial=DoorState.LOCKED)
door.allow_many(
    (DoorState.LOCKED, DoorState.CLOSED),
    (DoorState.CLOSED, DoorState.OPEN),
)
door.bind(id=42, topic=door_events)

door.transition(DoorState.CLOSED)
# → door #42: locked -> closed
door.transition(DoorState.OPEN)
# → door #42: closed -> open

One topic per state enum, many entities per topic — the id field on each StateChanged payload routes the event back to the right object. Subscribers are called in registration order; if you want a Bus keyed by event type, fsm_cub.events.Bus provides the @bus.on(EventType) shape too.

Putting it together — a small game

from enum import StrEnum

from fsm_cub import EventBoundFSM, StateChanged, Topic


class EnemyState(StrEnum):
    IDLE = "idle"
    ALERT = "alert"
    CHASING = "chasing"
    STUNNED = "stunned"
    DEAD = "dead"


enemy_events: Topic[StateChanged[EnemyState]] = Topic("enemies")


@enemy_events.subscribe
def on_state_change(ev: StateChanged[EnemyState]) -> None:
    if ev.to_state == EnemyState.ALERT:
        play_sound("alert_bark", id=ev.id)
    elif ev.to_state == EnemyState.DEAD:
        spawn_loot(id=ev.id)


def make_enemy(id: int) -> EventBoundFSM[EnemyState]:
    fsm = EventBoundFSM[EnemyState](initial=EnemyState.IDLE)
    fsm.allow_many(
        (EnemyState.IDLE, EnemyState.ALERT),
        (EnemyState.ALERT, EnemyState.CHASING),
        (EnemyState.ALERT, EnemyState.IDLE),
        (EnemyState.CHASING, EnemyState.STUNNED),
        (EnemyState.STUNNED, EnemyState.CHASING),
        (EnemyState.CHASING, EnemyState.DEAD),
        (EnemyState.STUNNED, EnemyState.DEAD),
    )
    fsm.bind(id=id, topic=enemy_events)
    return fsm

The FSM doesn't know audio or loot exist. The audio system doesn't know the FSM exists. They share only the StateChanged payload — that's the whole interface. Add another subscriber tomorrow (analytics, achievements, replay recording) and neither side changes.