State Machine Framework

A flexible state machine framework for Python with clean abstractions, ORM integration support, and type-safe workflow management.

Features

• Clean State Definitions: Minimal boilerplate with decorator-based validators and hooks
• Integration Agnostic: Works with any ORM via the adapter pattern
• Type Safety: Pydantic schema validation and type-checked workflow contexts
• External Registration: Register validators and hooks in separate modules for clean code organization
• Workflow Context: Type-safe context management with automatic validation

Installation

pip install state-machine-framework

Quick Start

1. Define Your Models

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Order(Base):
    __tablename__ = 'orders'

    id = Column(Integer, primary_key=True)
    order_id = Column(String(100), unique=True)
    status = Column(String(50))

2. Define a State Machine

from state_machine_framework import StateMachine

class OrderStateMachine(StateMachine):
    pass

OrderStateMachine.register(
    Order,
    state_field='status',
    identifier_field='order_id'
)

3. Define States

States declare which model field values they represent via model_states, and which states they can transition to via allowed_transitions.

from state_machine_framework import State

class OrderPending(State):
    _state_machine = OrderStateMachine
    is_start_state = True
    allowed_transitions = []  # filled in after all states are defined

    model_states = {Order: 'pending'}

class OrderProcessing(State):
    _state_machine = OrderStateMachine
    model_states = {Order: 'processing'}

class OrderCompleted(State):
    _state_machine = OrderStateMachine
    is_terminal_state = True
    model_states = {Order: 'completed'}

# Wire up the transition graph
OrderPending.allowed_transitions = [OrderProcessing]
OrderProcessing.allowed_transitions = [OrderCompleted]

4. Add Validators

Validators run after pre-transition hooks and before the state update. They should only raise exceptions — never mutate data.

from state_machine_framework import validator

# External registration (requires _state_machine to be set on the state)
@validator(OrderPending, order=1)
def validate_order_amount(state, data, context):
    if data.get('Order', {}).get('amount', 0) <= 0:
        raise ValueError("Order amount must be positive")
    return {'validated': True}

# Inline registration (inside the State class)
class OrderPending(State):
    @validator(order=1)
    def validate_order_amount(self, data, context):
        if data.get('Order', {}).get('amount', 0) <= 0:
            raise ValueError("Order amount must be positive")
        return {'validated': True}

5. Add Hooks

Hooks run after the transition completes. Use them for side effects such as sending notifications or updating external systems.

from state_machine_framework import hook

# External registration
@hook(OrderProcessing, order=1)
def send_confirmation_email(state, instances, context):
    order = instances['Order']
    send_email(order.customer_email)
    return {'email_sent': True}

# Inline registration
class OrderProcessing(State):
    @hook(order=1)
    def send_confirmation_email(self, instances, context):
        order = instances['Order']
        send_email(order.customer_email)
        return {'email_sent': True}

6. Add Pre-Transition Hooks

Pre-transition hooks run before validators. Use them to enrich or transform data or context (e.g. generating IDs, fetching external data).

from state_machine_framework import pre_transition
import uuid

@pre_transition(OrderPending, order=1)
def generate_order_id(state, data, context):
    data['Order']['order_id'] = f"ORD-{uuid.uuid4().hex[:8].upper()}"
    return {'id_generated': True}

7. Implement an ORM Adapter

from state_machine_framework import ORMAdapter
from state_machine_framework.core.exceptions import ObjectNotFound

class SQLAlchemyAdapter(ORMAdapter):
    def __init__(self, session):
        self.session = session

    def create(self, model_cls, **data):
        instance = model_cls(**data)
        self.session.add(instance)
        self.session.flush()
        return instance

    def get(self, model_cls, **filters):
        instance = self.session.query(model_cls).filter_by(**filters).one_or_none()
        if instance is None:
            raise ObjectNotFound(f"{model_cls.__name__} not found: {filters}")
        return instance

    def filter(self, model_cls, **filters):
        return self.session.query(model_cls).filter_by(**filters).all()

    def update(self, model_cls, filters, **updates):
        return self.session.query(model_cls).filter_by(**filters).update(updates)

    def delete(self, model_cls, **filters):
        return self.session.query(model_cls).filter_by(**filters).delete()

    def begin(self):    self.session.begin()
    def commit(self):   self.session.commit()
    def rollback(self): self.session.rollback()

8. Define a Workflow Context

from pydantic import BaseModel, Field
from state_machine_framework import WorkflowContext

class OrderData(BaseModel):
    amount: float = Field(..., gt=0)
    customer_email: str

class OrderWorkflowContext(WorkflowContext):
    def define_requirements(self):
        self.require_value('customer_id', str)
        self.require_model('order', Order, created_in_state=OrderPending)
        self.require_dict('order_data', OrderData)

9. Execute a Workflow

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite:///orders.db')
session = sessionmaker(bind=engine)()

adapter = SQLAlchemyAdapter(session)

with OrderWorkflowContext(
    OrderStateMachine,
    customer_id='CUST123',
    order_data={'amount': 99.99, 'customer_email': 'user@example.com'},
    _orm_adapter=adapter,
) as workflow:
    workflow.transition_to(OrderPending, Order={'amount': 99.99, 'customer_email': 'user@example.com'})
    workflow.transition_to(OrderProcessing)
    workflow.transition_to(OrderCompleted)

    print(workflow.get_history_summary())

Package Structure

state_machine_framework/
├── __init__.py              # Main package exports
├── core/
│   ├── __init__.py
│   ├── state.py             # State base class and metaclass
│   ├── state_machine.py     # StateMachine base class
│   └── exceptions.py        # Custom exceptions
├── decorators/
│   ├── __init__.py
│   └── hooks.py             # Decorators: pre_transition, validator, hook, transition
├── orm/
│   ├── __init__.py
│   └── base.py              # Abstract ORM adapter interface
└── workflow/
    ├── __init__.py
    ├── context.py           # WorkflowContext implementation
    └── requirements.py      # Context requirement classes

Key Concepts

States

States represent points in your workflow. Each state declares:

• model_states: maps model class → state field value at this state
• allowed_transitions: list of State classes this state can transition to (empty = unrestricted)
• is_start_state / is_terminal_state: lifecycle flags

class MyState(State):
    _state_machine = MyStateMachine
    is_start_state = True
    allowed_transitions = [NextState]
    model_states = {MyModel: 'my_value'}

Validators

Validators run after pre-transition hooks and before the state update. They receive (data, context) and must only raise exceptions — never modify data.

@validator(MyState, order=1)
def validate_something(state, data, context):
    if not some_condition(data):
        raise ValueError("Validation failed")
    return {'result': 'validated'}

Hooks

Hooks run after the transition completes. They receive (instances, context) where instances maps model name strings to ORM instances.

@hook(MyState, order=1)
def do_something(state, instances, context):
    my_model = instances['MyModel']
    notify(my_model)
    return {'action': 'completed'}

Pre-Transition Hooks

Pre-transition hooks run before validators. They receive (data, context) and are allowed to mutate both, making them suitable for data enrichment.

@pre_transition(MyState, order=1)
def enrich_data(state, data, context):
    data['MyModel']['computed_field'] = compute_value(data)
    return {'enriched': True}

Workflow Context

Workflow contexts provide type-safe, validated context management:

class MyWorkflowContext(WorkflowContext):
    def define_requirements(self):
        # Require a simple typed value
        self.require_value('user_id', str, required=True)

        # Require a model instance (created during workflow at the given state)
        self.require_model('order', Order, created_in_state=OrderPending)

        # Require a dict validated against a Pydantic schema
        self.require_dict('order_data', OrderDataSchema, required=True)

Custom Transitions

Use @transition to override the default state-field update logic:

class MyState(State):
    @transition
    def apply(self, instances, context):
        for instance in instances.values():
            instance.status = 'custom'
            instance.save()

Examples

See the examples/ directory for a complete vending machine implementation demonstrating:

• ORM adapter implementation
• Workflow context with validation
• External validator and hook registration
• Pydantic schema validation
• Complete transaction workflow

To run the example:

cd examples/vending_machine
python main.py

Requirements

• Python 3.10+
• pydantic >= 2.0

License

MIT License - see LICENSE file for details

Contributing

Contributions are welcome! Please feel free to reach out or open an issue.