Python State Machines with Transitions — Deep Dive

Advanced transitions patterns — hierarchical states, async machines, persistence, diagram generation, and integrating state machines with Django and FastAPI.

Full Working Example

from transitions import Machine

class Order:
    states = ["pending", "paid", "shipped", "delivered", "cancelled", "refunded"]

    transitions = [
        {"trigger": "pay",      "source": "pending",   "dest": "paid",
         "conditions": ["is_payment_valid"], "after": "send_receipt"},
        {"trigger": "ship",     "source": "paid",      "dest": "shipped",
         "after": "notify_shipping"},
        {"trigger": "deliver",  "source": "shipped",   "dest": "delivered",
         "after": "request_review"},
        {"trigger": "cancel",   "source": ["pending", "paid"], "dest": "cancelled",
         "before": "check_cancellation_window"},
        {"trigger": "refund",   "source": ["delivered", "shipped"], "dest": "refunded",
         "conditions": ["within_refund_period"], "after": "process_refund"},
    ]

    def __init__(self, order_id: str, total: float):
        self.order_id = order_id
        self.total = total
        self.machine = Machine(
            model=self,
            states=Order.states,
            transitions=Order.transitions,
            initial="pending",
            send_event=True,  # pass EventData to callbacks
        )

    def is_payment_valid(self, event):
        return event.kwargs.get("amount", 0) >= self.total

    def within_refund_period(self, event):
        from datetime import datetime, timedelta
        delivered_at = getattr(self, "delivered_at", None)
        if not delivered_at:
            return True
        return datetime.now() - delivered_at < timedelta(days=30)

    def send_receipt(self, event):
        print(f"Receipt sent for order {self.order_id}")

    def notify_shipping(self, event):
        print(f"Order {self.order_id} shipped")

    def request_review(self, event):
        from datetime import datetime
        self.delivered_at = datetime.now()
        print(f"Review requested for order {self.order_id}")

    def check_cancellation_window(self, event):
        # Could raise an exception to abort the transition
        pass

    def process_refund(self, event):
        print(f"Refund of ${self.total} processed for order {self.order_id}")

# Usage
order = Order("ORD-001", 59.99)
print(order.state)           # "pending"
print(order.may_pay())       # True
print(order.may_ship())      # False — not paid yet

order.pay(amount=59.99)      # → paid
order.ship()                 # → shipped
order.deliver()              # → delivered
print(order.is_delivered())  # True

Hierarchical (Nested) State Machines

The HierarchicalMachine extension supports nested states with NestedState:

from transitions.extensions import HierarchicalMachine

states = [
    "idle",
    {
        "name": "active",
        "children": [
            "processing",
            "waiting_input",
            {
                "name": "error",
                "children": ["recoverable", "fatal"],
            },
        ],
    },
    "shutdown",
]

transitions = [
    {"trigger": "start",       "source": "idle",                        "dest": "active_processing"},
    {"trigger": "need_input",  "source": "active_processing",           "dest": "active_waiting_input"},
    {"trigger": "resume",      "source": "active_waiting_input",        "dest": "active_processing"},
    {"trigger": "fail",        "source": "active_processing",           "dest": "active_error_recoverable"},
    {"trigger": "escalate",    "source": "active_error_recoverable",    "dest": "active_error_fatal"},
    {"trigger": "recover",     "source": "active_error_recoverable",    "dest": "active_processing"},
    {"trigger": "stop",        "source": "active",                      "dest": "shutdown"},  # exits from any child
]

machine = HierarchicalMachine(
    states=states,
    transitions=transitions,
    initial="idle",
)

Nested states let you define transitions on parent states that apply to all children — stop from active works whether you’re in active_processing, active_error_recoverable, or any other child state.

Async State Machines

For asyncio-based applications:

from transitions.extensions.asyncio import AsyncMachine

class AsyncOrder:
    states = ["pending", "processing", "complete", "failed"]

    def __init__(self):
        self.machine = AsyncMachine(
            model=self,
            states=self.states,
            initial="pending",
            transitions=[
                {"trigger": "process", "source": "pending",    "dest": "processing",
                 "after": "run_processing"},
                {"trigger": "complete", "source": "processing", "dest": "complete"},
                {"trigger": "fail",     "source": "processing", "dest": "failed"},
            ],
        )

    async def run_processing(self):
        import asyncio
        await asyncio.sleep(1)  # simulate async work
        print("Processing complete")

# Usage
import asyncio

async def main():
    order = AsyncOrder()
    await order.process()  # triggers are now awaitable
    print(order.state)     # "processing"

asyncio.run(main())

Persistence: Saving and Restoring State

State machines are in-memory by default. For persistence, serialize the state:

import json

class PersistentOrder(Order):
    def save(self) -> str:
        return json.dumps({
            "order_id": self.order_id,
            "total": self.total,
            "state": self.state,
        })

    @classmethod
    def load(cls, data: str) -> "PersistentOrder":
        d = json.loads(data)
        order = cls(d["order_id"], d["total"])
        # Force state without triggering transitions
        order.machine.set_state(d["state"])
        return order

Django Integration

from django.db import models
from transitions import Machine

class OrderModel(models.Model):
    STATES = ["pending", "paid", "shipped", "delivered", "cancelled"]

    order_id = models.CharField(max_length=50, unique=True)
    total = models.DecimalField(max_digits=10, decimal_places=2)
    state = models.CharField(max_length=20, default="pending")

    class Meta:
        db_table = "orders"

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.machine = Machine(
            model=self,
            states=self.STATES,
            initial=self.state,
            transitions=[
                {"trigger": "pay",  "source": "pending", "dest": "paid",
                 "after": "save_state"},
                {"trigger": "ship", "source": "paid",    "dest": "shipped",
                 "after": "save_state"},
            ],
            model_attribute="state",
        )

    def save_state(self):
        self.save(update_fields=["state"])

Diagram Generation

transitions can generate state diagrams using Graphviz:

from transitions.extensions import GraphMachine

class VisualOrder:
    pass

machine = GraphMachine(
    model=VisualOrder(),
    states=["pending", "paid", "shipped", "delivered"],
    transitions=[
        {"trigger": "pay",     "source": "pending", "dest": "paid"},
        {"trigger": "ship",    "source": "paid",    "dest": "shipped"},
        {"trigger": "deliver", "source": "shipped", "dest": "delivered"},
    ],
    initial="pending",
    show_conditions=True,
    show_state_attributes=True,
)

# Generate PNG
machine.get_graph().draw("order_workflow.png", prog="dot")

This is invaluable for documentation — the diagram is always in sync with the code because it’s generated from the same transition definitions.

Thread Safety

The default Machine is not thread-safe. For concurrent access:

from transitions.extensions import LockedMachine

machine = LockedMachine(
    model=my_model,
    states=states,
    transitions=transitions,
    initial="idle",
)
# All trigger calls are now mutex-protected

LockedMachine wraps each trigger call in a threading.Lock. For high-contention scenarios, consider per-model locking or event queuing instead.

Custom Transition Validation

Beyond simple conditions, you can implement complex validation by overriding transition behavior:

from transitions import Machine, MachineError

class StrictOrder:
    def __init__(self):
        self.approved_by = None
        self.amount = 0
        self.machine = Machine(
            model=self,
            states=["draft", "pending_approval", "approved", "executed"],
            transitions=[
                {
                    "trigger": "submit",
                    "source": "draft",
                    "dest": "pending_approval",
                    "conditions": ["has_valid_amount"],
                    "unless": ["is_duplicate"],
                },
                {
                    "trigger": "approve",
                    "source": "pending_approval",
                    "dest": "approved",
                    "conditions": ["has_approver", "approver_has_authority"],
                },
            ],
            initial="draft",
            send_event=True,
        )

    def has_valid_amount(self, event):
        return 0 < self.amount <= 1_000_000

    def is_duplicate(self, event):
        return False  # check against database

    def has_approver(self, event):
        return event.kwargs.get("approver") is not None

    def approver_has_authority(self, event):
        approver = event.kwargs.get("approver")
        if self.amount > 10_000:
            return approver.role == "director"
        return approver.role in ("manager", "director")

The unless parameter inverts conditions — the transition proceeds only if is_duplicate returns False.

Testing State Machines

import pytest
from transitions import MachineError

def test_valid_order_flow():
    order = Order("TEST-001", 29.99)
    assert order.state == "pending"

    order.pay(amount=29.99)
    assert order.state == "paid"

    order.ship()
    assert order.state == "shipped"

def test_cannot_ship_unpaid_order():
    order = Order("TEST-002", 29.99)
    with pytest.raises(MachineError):
        order.ship()  # Can't ship from "pending"

def test_insufficient_payment_blocked():
    order = Order("TEST-003", 100.00)
    order.pay(amount=50.00)  # Condition fails
    assert order.state == "pending"  # State unchanged

def test_may_checks():
    order = Order("TEST-004", 29.99)
    assert order.may_pay() is True
    assert order.may_ship() is False

    order.pay(amount=29.99)
    assert order.may_pay() is False
    assert order.may_ship() is True

Alternatives to transitions

LibraryApproachBest For
transitionsClass-based, mixin-styleGeneral-purpose, most popular
python-statemachineDeclarative class syntaxClean API, good docs
sismicStatechart interpreter (SCXML)Complex hierarchical machines
xstate-pythonPort of XStateJavaScript ecosystem familiarity
enum + matchDIY with Python 3.10+Minimal dependencies

Performance Notes

  • Transition dispatch: ~50,000 transitions/sec (with callbacks)
  • Without callbacks: ~200,000 transitions/sec
  • Memory per machine: ~2-5 KB depending on state/transition count
  • The overhead is negligible for business logic workflows but inappropriate for per-packet or per-frame processing

One thing to remember: The transitions library turns implicit state logic (scattered if/else chains) into explicit, testable, visualizable workflow definitions — and its extensions for async, hierarchy, locking, and diagram generation cover most production needs without leaving the ecosystem.

pythonstate-machinestransitions

See Also

  • Python Event Emitter Patterns How Python programs shout 'something happened!' so other parts of the code can react — like a school bell that tells everyone it's recess.
  • Python Observer Vs Pubsub Two ways Python code can share news — one is like telling your friends directly, the other is like posting on a bulletin board for anyone to read.
  • Python Rxpy Reactive Programming How RxPY lets Python code react to streams of data the way a news ticker reacts to breaking stories — automatically and in real time.
  • Ci Cd Why big apps can ship updates every day without turning your phone into a glitchy mess — CI/CD is the behind-the-scenes quality gate and delivery truck.
  • Containerization Why does software that works on your computer break on everyone else's? Containers fix that — and they're why Netflix can deploy 100 updates a day without the site going down.

Related Topics

ab-testing How tech companies run thousands of experiments at once to improve their products — the scientific method applied to everything from button colors to recommendation algorithms. 3 levels → activation-functions Why neural networks need these tiny mathematical functions — and how ReLU's simplicity accidentally made deep learning possible. 3 levels → ai-agents-architecture How AI systems go from answering questions to actually doing things — the design patterns that turn language models into autonomous agents that browse, code, and plan. 3 levels →