0
# Mixins and Integration
1
2
Domain model integration patterns, mixin classes for automatic state machine binding, and framework integration including Django support and model field binding.
3
4
## Capabilities
5
6
### MachineMixin Class
7
8
Mixin class that enables models to automatically instantiate and manage state machines with configurable binding options.
9
10
```python { .api }
11
class MachineMixin:
12
"""
13
Mixin that allows a model to automatically instantiate and assign a StateMachine.
14
15
This mixin provides seamless integration between domain models and state machines,
16
automatically handling state machine instantiation and event binding.
17
18
Class Attributes:
19
- state_field_name: The model's field name that holds the state value (default: "state")
20
- state_machine_name: Fully qualified name of the StateMachine class for import
21
- state_machine_attr: Name of the model attribute that will hold the machine instance (default: "statemachine")
22
- bind_events_as_methods: If True, state machine events are bound as model methods (default: False)
23
"""
24
25
state_field_name: str = "state"
26
"""The model's state field name that will hold the state value."""
27
28
state_machine_name: str = None
29
"""A fully qualified name of the StateMachine class, where it can be imported."""
30
31
state_machine_attr: str = "statemachine"
32
"""Name of the model's attribute that will hold the machine instance."""
33
34
bind_events_as_methods: bool = False
35
"""If True, state machine event triggers will be bound to the model as methods."""
36
37
def __init__(self, *args, **kwargs):
38
"""
39
Initialize the mixin and create state machine instance.
40
41
Raises:
42
- ValueError: If state_machine_name is not set or invalid
43
"""
44
```
45
46
### Registry Functions
47
48
Functions for state machine registration and discovery, particularly useful for framework integration.
49
50
```python { .api }
51
# Registry module functions for state machine discovery
52
def get_machine_cls(qualname: str):
53
"""
54
Get state machine class by fully qualified name.
55
56
Parameters:
57
- qualname: Fully qualified class name (e.g., "myapp.machines.OrderMachine")
58
59
Returns:
60
StateMachine class
61
62
Raises:
63
- ImportError: If the module or class cannot be imported
64
"""
65
66
def register_machine(machine_cls):
67
"""
68
Register a state machine class for discovery.
69
70
Parameters:
71
- machine_cls: StateMachine class to register
72
"""
73
```
74
75
### Model Integration Utilities
76
77
Utilities for integrating state machines with various model systems and frameworks.
78
79
```python { .api }
80
def bind_events_to(machine: StateMachine, target_object: object):
81
"""
82
Bind state machine events as methods on target object.
83
84
Parameters:
85
- machine: StateMachine instance
86
- target_object: Object to bind events to
87
88
Creates methods on target_object for each event in the machine.
89
"""
90
91
class Model:
92
"""
93
Base model class with state machine integration support.
94
95
Provides foundation for external model integration with
96
automatic state field management.
97
"""
98
def __init__(self, **kwargs):
99
"""Initialize model with optional state field."""
100
```
101
102
## Usage Examples
103
104
### Basic Model Integration with MachineMixin
105
106
```python
107
from statemachine import StateMachine, State
108
from statemachine.mixins import MachineMixin
109
110
# Define the state machine
111
class OrderStateMachine(StateMachine):
112
pending = State(initial=True, value="pending")
113
paid = State(value="paid")
114
shipped = State(value="shipped")
115
delivered = State(value="delivered", final=True)
116
cancelled = State(value="cancelled", final=True)
117
118
pay = pending.to(paid)
119
ship = paid.to(shipped)
120
deliver = shipped.to(delivered)
121
cancel = (
122
pending.to(cancelled)
123
| paid.to(cancelled)
124
| shipped.to(cancelled)
125
)
126
127
def on_enter_paid(self, payment_info: dict = None):
128
print(f"Payment received: {payment_info}")
129
130
def on_enter_shipped(self, tracking_number: str = None):
131
print(f"Order shipped with tracking: {tracking_number}")
132
133
def on_enter_delivered(self):
134
print("Order delivered successfully!")
135
136
# Define the model using MachineMixin
137
class Order(MachineMixin):
138
state_machine_name = "OrderStateMachine" # Could be full path like "myapp.machines.OrderStateMachine"
139
state_field_name = "status"
140
bind_events_as_methods = True
141
142
def __init__(self, order_id: str, **kwargs):
143
self.order_id = order_id
144
self.status = "pending" # Initial state value
145
self.customer_email = kwargs.get("customer_email")
146
super().__init__(**kwargs) # Initialize MachineMixin
147
148
def send_notification(self, message: str):
149
"""Custom model method."""
150
print(f"Notification to {self.customer_email}: {message}")
151
152
# Usage
153
order = Order("ORD-001", customer_email="customer@example.com")
154
155
# Access state machine via configured attribute
156
print(f"Current state: {order.statemachine.current_state.id}") # "pending"
157
print(f"Model status field: {order.status}") # "pending"
158
159
# Use bound event methods (since bind_events_as_methods=True)
160
order.pay(payment_info={"method": "credit_card", "amount": 99.99})
161
print(f"After payment - Status: {order.status}") # "paid"
162
163
order.ship(tracking_number="TRK123456")
164
print(f"After shipping - Status: {order.status}") # "shipped"
165
166
order.deliver()
167
print(f"Final status: {order.status}") # "delivered"
168
```
169
170
### Advanced Model Integration with Custom State Field
171
172
```python
173
import json
174
from datetime import datetime
175
176
class AdvancedOrder(MachineMixin):
177
state_machine_name = "OrderStateMachine"
178
state_field_name = "workflow_state"
179
state_machine_attr = "workflow"
180
181
def __init__(self, order_id: str, **kwargs):
182
self.order_id = order_id
183
self.workflow_state = "pending"
184
self.created_at = datetime.now()
185
self.state_history = []
186
super().__init__(**kwargs)
187
188
# Add state change logging
189
self.workflow.add_callback(
190
self.log_state_change,
191
group=CallbackGroup.ACTIONS
192
)
193
194
def log_state_change(self, event: str, source: State, target: State, **kwargs):
195
"""Log all state changes."""
196
self.state_history.append({
197
"timestamp": datetime.now().isoformat(),
198
"event": event,
199
"from": source.id,
200
"to": target.id,
201
"metadata": kwargs
202
})
203
print(f"State change logged: {source.id} -> {target.id} via {event}")
204
205
def get_state_history(self) -> str:
206
"""Get formatted state history."""
207
return json.dumps(self.state_history, indent=2)
208
209
def current_state_info(self) -> dict:
210
"""Get current state information."""
211
return {
212
"state": self.workflow_state,
213
"state_name": self.workflow.current_state.name,
214
"is_final": self.workflow.current_state.final,
215
"allowed_events": [e.id for e in self.workflow.allowed_events]
216
}
217
218
# Usage
219
order = AdvancedOrder("ORD-002")
220
print("Initial state:", order.current_state_info())
221
222
order.workflow.send("pay", payment_method="paypal", amount=75.50)
223
order.workflow.send("ship", carrier="UPS", tracking="1Z999AA1234567890")
224
order.workflow.send("deliver")
225
226
print("\nFinal state:", order.current_state_info())
227
print("\nState history:")
228
print(order.get_state_history())
229
```
230
231
### Django Model Integration
232
233
```python
234
# Django model example (requires Django)
235
try:
236
from django.db import models
237
from statemachine.mixins import MachineMixin
238
239
class DjangoOrder(models.Model, MachineMixin):
240
# Django model fields
241
order_id = models.CharField(max_length=50, unique=True)
242
customer_email = models.EmailField()
243
total_amount = models.DecimalField(max_digits=10, decimal_places=2)
244
status = models.CharField(max_length=20, default="pending")
245
created_at = models.DateTimeField(auto_now_add=True)
246
updated_at = models.DateTimeField(auto_now=True)
247
248
# MachineMixin configuration
249
state_machine_name = "myapp.machines.OrderStateMachine"
250
state_field_name = "status"
251
bind_events_as_methods = True
252
253
class Meta:
254
db_table = "orders"
255
256
def save(self, *args, **kwargs):
257
"""Override save to sync state machine state."""
258
if hasattr(self, 'statemachine'):
259
# Ensure model field matches state machine state
260
self.status = self.statemachine.current_state.value
261
super().save(*args, **kwargs)
262
263
def __str__(self):
264
return f"Order {self.order_id} - {self.status}"
265
266
# Usage in Django views/services
267
def process_payment(order_id: str, payment_data: dict):
268
"""Service function to process order payment."""
269
order = DjangoOrder.objects.get(order_id=order_id)
270
271
try:
272
# Use bound event method
273
order.pay(payment_info=payment_data)
274
order.save() # Persist state change
275
return {"success": True, "new_state": order.status}
276
except TransitionNotAllowed as e:
277
return {"success": False, "error": str(e)}
278
279
except ImportError:
280
print("Django not available - skipping Django example")
281
```
282
283
### SQLAlchemy Integration
284
285
```python
286
# SQLAlchemy model example
287
try:
288
from sqlalchemy import Column, Integer, String, DateTime, create_engine
289
from sqlalchemy.ext.declarative import declarative_base
290
from sqlalchemy.orm import sessionmaker
291
from datetime import datetime
292
293
Base = declarative_base()
294
295
class SQLAlchemyOrder(Base, MachineMixin):
296
__tablename__ = "orders"
297
298
# SQLAlchemy columns
299
id = Column(Integer, primary_key=True)
300
order_id = Column(String(50), unique=True, nullable=False)
301
customer_email = Column(String(255), nullable=False)
302
status = Column(String(20), default="pending")
303
created_at = Column(DateTime, default=datetime.now)
304
updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
305
306
# MachineMixin configuration
307
state_machine_name = "OrderStateMachine"
308
state_field_name = "status"
309
310
def __init__(self, order_id: str, customer_email: str, **kwargs):
311
self.order_id = order_id
312
self.customer_email = customer_email
313
super().__init__(**kwargs)
314
315
def update_state(self, session):
316
"""Helper to update and persist state changes."""
317
if hasattr(self, 'statemachine'):
318
self.status = self.statemachine.current_state.value
319
self.updated_at = datetime.now()
320
session.commit()
321
322
# Usage
323
def create_order_with_workflow():
324
engine = create_engine("sqlite:///orders.db")
325
Base.metadata.create_all(engine)
326
Session = sessionmaker(bind=engine)
327
session = Session()
328
329
# Create new order
330
order = SQLAlchemyOrder(
331
order_id="ORD-003",
332
customer_email="customer@example.com"
333
)
334
session.add(order)
335
session.commit()
336
337
# Process through workflow
338
order.statemachine.send("pay", payment_method="stripe")
339
order.update_state(session)
340
341
order.statemachine.send("ship", carrier="FedEx")
342
order.update_state(session)
343
344
print(f"Order {order.order_id} status: {order.status}")
345
session.close()
346
347
except ImportError:
348
print("SQLAlchemy not available - skipping SQLAlchemy example")
349
```
350
351
### Custom Integration Pattern
352
353
```python
354
from abc import ABC, abstractmethod
355
356
class StateMachineModel(ABC):
357
"""Abstract base class for state machine model integration."""
358
359
def __init__(self, machine_class, initial_state_value=None, **kwargs):
360
self.machine_class = machine_class
361
self._state_value = initial_state_value or self.get_initial_state_value()
362
self._machine = None
363
self.initialize_machine(**kwargs)
364
365
@abstractmethod
366
def get_initial_state_value(self):
367
"""Get the initial state value for this model."""
368
pass
369
370
@abstractmethod
371
def persist_state(self):
372
"""Persist the current state to storage."""
373
pass
374
375
def initialize_machine(self, **kwargs):
376
"""Initialize the state machine instance."""
377
self._machine = self.machine_class(
378
model=self,
379
state_field="_state_value",
380
**kwargs
381
)
382
383
@property
384
def machine(self):
385
"""Get the state machine instance."""
386
return self._machine
387
388
def send_event(self, event: str, *args, **kwargs):
389
"""Send event and persist state change."""
390
try:
391
result = self._machine.send(event, *args, **kwargs)
392
self.persist_state()
393
return result
394
except Exception as e:
395
# Optionally log error or handle rollback
396
raise
397
398
class FileBasedOrder(StateMachineModel):
399
"""Example model that persists state to file."""
400
401
def __init__(self, order_id: str, **kwargs):
402
self.order_id = order_id
403
self.filename = f"order_{order_id}.json"
404
super().__init__(OrderStateMachine, **kwargs)
405
406
def get_initial_state_value(self):
407
"""Load state from file or return default."""
408
try:
409
with open(self.filename, 'r') as f:
410
data = json.load(f)
411
return data.get('state', 'pending')
412
except FileNotFoundError:
413
return 'pending'
414
415
def persist_state(self):
416
"""Save current state to file."""
417
data = {
418
'order_id': self.order_id,
419
'state': self._state_value,
420
'timestamp': datetime.now().isoformat()
421
}
422
with open(self.filename, 'w') as f:
423
json.dump(data, f)
424
print(f"State persisted: {self._state_value}")
425
426
# Usage
427
order = FileBasedOrder("ORD-004")
428
print(f"Initial state: {order.machine.current_state.id}")
429
430
order.send_event("pay", payment_method="bitcoin")
431
order.send_event("ship", carrier="DHL")
432
order.send_event("deliver")
433
434
print(f"Final state: {order.machine.current_state.id}")
435
```
436
437
### Event Binding Patterns
438
439
```python
440
class EventBoundModel(MachineMixin):
441
state_machine_name = "OrderStateMachine"
442
bind_events_as_methods = True
443
444
def __init__(self, order_id: str, **kwargs):
445
self.order_id = order_id
446
self.state = "pending"
447
self.notifications = []
448
super().__init__(**kwargs)
449
450
# Add custom event handling
451
self.setup_event_notifications()
452
453
def setup_event_notifications(self):
454
"""Setup automatic notifications for events."""
455
# Override bound event methods to add notifications
456
original_pay = self.pay
457
original_ship = self.ship
458
original_deliver = self.deliver
459
original_cancel = self.cancel
460
461
def notify_pay(*args, **kwargs):
462
result = original_pay(*args, **kwargs)
463
self.notifications.append("Payment confirmation sent")
464
return result
465
466
def notify_ship(*args, **kwargs):
467
result = original_ship(*args, **kwargs)
468
self.notifications.append("Shipping notification sent")
469
return result
470
471
def notify_deliver(*args, **kwargs):
472
result = original_deliver(*args, **kwargs)
473
self.notifications.append("Delivery confirmation sent")
474
return result
475
476
def notify_cancel(*args, **kwargs):
477
result = original_cancel(*args, **kwargs)
478
self.notifications.append("Cancellation notice sent")
479
return result
480
481
# Replace bound methods with notification versions
482
self.pay = notify_pay
483
self.ship = notify_ship
484
self.deliver = notify_deliver
485
self.cancel = notify_cancel
486
487
def get_notifications(self):
488
"""Get all notifications sent."""
489
return self.notifications
490
491
# Usage
492
order = EventBoundModel("ORD-005")
493
order.pay(payment_method="visa")
494
order.ship(tracking_number="ABC123")
495
order.deliver()
496
497
print("Notifications sent:")
498
for notification in order.get_notifications():
499
print(f"- {notification}")
500
```