state-machine-framework 1.0.0

A flexible state machine framework for Python

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 ORM via 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 States

from state_machine_framework import State

class OrderPending(State):
    is_start_state = True
    
    class Meta:
        order = 1
        Order_status = 'pending'

class OrderProcessing(State):
    class Meta:
        order = 2
        Order_status = 'processing'

class OrderCompleted(State):
    is_terminal_state = True
    
    class Meta:
        order = 3
        Order_status = 'completed'

3. Define State Machine

from state_machine_framework import StateMachine

class OrderStateMachine(StateMachine):
    pass

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

4. Add Validators (External Registration)

from state_machine_framework import validator

@validator(OrderPending, order=1)
def validate_order_amount(state, data, context):
    if data.get('amount', 0) <= 0:
        raise ValueError("Invalid amount")
    return {'validated': True}

5. Add Hooks (External Registration)

from state_machine_framework import hook

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

6. Define Pydantic Schemas

from pydantic import BaseModel, Field

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

7. Define Workflow Context

from state_machine_framework import WorkflowContext

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)

8. Execute Workflow

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

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

with OrderWorkflowContext(
    OrderStateMachine,
    customer_id='CUST123',
    order_data={'amount': 99.99, 'customer_email': '[e-mail]'},
    _session=session
) as workflow:
    
    workflow.transition_to(OrderPending)
    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 for validators, hooks, transitions
├── orm/
│   ├── __init__.py
│   ├── base.py              # Abstract ORM adapter
│   └── sqlalchemy_adapter.py # SQLAlchemy implementation
└── workflow/
    ├── __init__.py
    ├── context.py           # WorkflowContext implementation
    └── requirements.py      # Context requirement classes

Key Concepts

States

States represent points in your workflow. Define them by subclassing State:

class MyState(State):
    is_start_state = True  # Optional: mark as initial state
    is_terminal_state = False  # Optional: mark as final state
    
    class Meta:
        order = 1  # Execution order
        ModelName_status = 'state_value'  # State value for model

Validators

Validators run before state transitions and perform validation logic:

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

Hooks

Hooks run after state transitions and perform side effects:

@hook(MyState, order=1)
def do_something(state, data, context):
    # Side effects here (logging, notifications, etc.)
    instances = data['instances']  # Access model instances
    return {'action': 'completed'}

Workflow Context

Workflow contexts provide type-safe, validated context management:

class MyWorkflowContext(WorkflowContext):
    def define_requirements(self):
        # Require a simple value
        self.require_value('user_id', str, required=True)
        
        # Require a model instance (may be created during workflow)
        self.require_model('order', Order, created_in_state=OrderPending)
        
        # Require a dict validated by Pydantic schema
        self.require_dict('order_data', OrderDataSchema, required=True)

Examples

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

• SQLAlchemy ORM integration
• Workflow context with validation
• External validator and hook registration
• Type-safe context management
• Pydantic schema validation
• Complete transaction workflow

To run the example:

cd examples/vending_machine
python main.py

Dynamic ORM Adapters

Implement the ORMAdapter interface for your ORM:

from state_machine_framework import ORMAdapter, ORMAdapterFactory

class MyORMAdapter(ORMAdapter):
    def create(self, model_cls, **data):
        # Your implementation
        pass
    
    # ... implement all abstract methods

# Register it
ORMAdapterFactory.register_adapter('myorm', MyORMAdapter)

# Use it
sm = MyStateMachine(data, using_orm='myorm')

Requirements

• Python 3.8+
• pydantic >= 1.8.0

License

MIT License - see LICENSE file for details

Contributing

Contributions are welcome! Please feel free to reach out...