0
# Utilities and Data Classes
1
2
Additional utility classes, data structures, and helper functions that support state machine operations including event data handling, model integration, and internal state management.
3
4
## Capabilities
5
6
### Event Data Classes
7
8
Data structures used internally for event processing and callback parameter injection.
9
10
```python { .api }
11
from dataclasses import dataclass
12
from typing import Any
13
14
@dataclass
15
class TriggerData:
16
"""
17
Container for event trigger information passed through the state machine.
18
19
Attributes:
20
- machine: StateMachine instance processing the event
21
- event: Event object that was triggered
22
- model: External model object (if provided)
23
- args: Positional arguments passed to the event
24
- kwargs: Keyword arguments passed to the event
25
"""
26
machine: StateMachine
27
event: Event
28
model: Any = field(init=False)
29
args: tuple = field(default_factory=tuple)
30
kwargs: dict = field(default_factory=dict)
31
32
@dataclass
33
class EventData:
34
"""
35
Complete event processing context containing trigger and transition information.
36
37
Attributes:
38
- trigger_data: TriggerData instance with event information
39
- transition: Transition object being executed
40
- state: Current state (computed from trigger_data)
41
- source: Source state of transition (computed from transition)
42
- target: Target state of transition (computed from transition)
43
- result: Result value from callback execution
44
- executed: Whether the transition has been executed
45
"""
46
trigger_data: TriggerData
47
transition: Transition
48
state: State = field(init=False)
49
source: State = field(init=False)
50
target: State = field(init=False)
51
result: Any | None = None
52
executed: bool = False
53
```
54
55
### Events Collection
56
57
Container class for managing multiple Event objects within a state machine.
58
59
```python { .api }
60
class Events:
61
"""
62
Collection class for managing Event objects in a state machine.
63
64
Provides methods for adding events and matching event identifiers.
65
"""
66
def __init__(self): ...
67
68
def add(self, events):
69
"""
70
Add events to the collection.
71
72
Parameters:
73
- events: Event object or iterable of Event objects
74
"""
75
76
def match(self, event: str) -> bool:
77
"""
78
Check if any event in collection matches the given event identifier.
79
80
Parameters:
81
- event: Event identifier string
82
83
Returns:
84
True if any event matches, False otherwise
85
"""
86
87
def __iter__(self):
88
"""Iterate over events in the collection."""
89
90
def __len__(self) -> int:
91
"""Get number of events in the collection."""
92
```
93
94
### Model Integration
95
96
Base model class for external state storage and integration patterns.
97
98
```python { .api }
99
class Model:
100
"""
101
Simple model class for external state storage.
102
103
Provides a basic foundation for state machine model integration
104
with automatic state field management.
105
"""
106
def __init__(self):
107
"""Initialize model with state field."""
108
self.state = None
109
```
110
111
### Registry Functions
112
113
Functions for state machine class registration and discovery, useful for dynamic loading and framework integration.
114
115
```python { .api }
116
def register(cls):
117
"""
118
Register a StateMachine class for discovery.
119
120
Parameters:
121
- cls: StateMachine class to register
122
123
Returns:
124
The registered class (allows use as decorator)
125
"""
126
127
def get_machine_cls(name: str):
128
"""
129
Get registered StateMachine class by name.
130
131
Parameters:
132
- name: Fully qualified class name
133
134
Returns:
135
StateMachine class
136
137
Raises:
138
ImportError: If class is not found or cannot be imported
139
"""
140
141
def init_registry():
142
"""Initialize the state machine registry."""
143
144
def load_modules(modules=None):
145
"""
146
Load modules containing state machine definitions.
147
148
Parameters:
149
- modules: List of module names to load (optional)
150
"""
151
```
152
153
### Utility Functions
154
155
Helper functions for common operations and type handling.
156
157
```python { .api }
158
def qualname(cls) -> str:
159
"""
160
Get fully qualified name of a class.
161
162
Parameters:
163
- cls: Class object
164
165
Returns:
166
Fully qualified class name string
167
"""
168
169
def ensure_iterable(obj):
170
"""
171
Ensure object is iterable, wrapping single items in a tuple.
172
173
Parameters:
174
- obj: Object to make iterable
175
176
Returns:
177
Iterable version of the object
178
"""
179
180
def run_async_from_sync(coroutine):
181
"""
182
Run async coroutine from synchronous context.
183
184
Parameters:
185
- coroutine: Async coroutine to execute
186
187
Returns:
188
Result of coroutine execution
189
"""
190
```
191
192
## Usage Examples
193
194
### Event Data Access in Callbacks
195
196
```python
197
from statemachine import StateMachine, State
198
199
class ProcessingMachine(StateMachine):
200
idle = State(initial=True)
201
working = State()
202
done = State(final=True)
203
204
start = idle.to(working)
205
finish = working.to(done)
206
207
def on_start(self, event_data, trigger_data, **kwargs):
208
"""Access event data structures in callbacks."""
209
print(f"Event: {trigger_data.event.id}")
210
print(f"Machine: {trigger_data.machine.__class__.__name__}")
211
print(f"Transition: {event_data.transition.source.id} -> {event_data.transition.target.id}")
212
print(f"Args: {trigger_data.args}")
213
print(f"Kwargs: {trigger_data.kwargs}")
214
215
# Usage
216
machine = ProcessingMachine()
217
machine.send("start", priority="high", task_id="T001")
218
```
219
220
### Model Integration with External Objects
221
222
```python
223
from statemachine.model import Model
224
225
class OrderModel(Model):
226
def __init__(self, order_id: str):
227
super().__init__()
228
self.order_id = order_id
229
self.state = "pending" # Initial state value
230
self.created_at = datetime.now()
231
232
class OrderMachine(StateMachine):
233
pending = State("Pending", initial=True, value="pending")
234
paid = State("Paid", value="paid")
235
shipped = State("Shipped", value="shipped")
236
delivered = State("Delivered", final=True, value="delivered")
237
238
pay = pending.to(paid)
239
ship = paid.to(shipped)
240
deliver = shipped.to(delivered)
241
242
# Integration
243
order_model = OrderModel("ORD-001")
244
machine = OrderMachine(order_model)
245
246
# State is automatically synced with model
247
machine.send("pay")
248
print(order_model.state) # "paid"
249
```
250
251
### Registry Usage for Dynamic Loading
252
253
```python
254
from statemachine.registry import register, get_machine_cls
255
256
@register
257
class WorkflowMachine(StateMachine):
258
start = State(initial=True)
259
end = State(final=True)
260
261
proceed = start.to(end)
262
263
# Later, dynamically load the class
264
machine_cls = get_machine_cls("__main__.WorkflowMachine")
265
instance = machine_cls()
266
print(f"Loaded: {instance.__class__.__name__}")
267
```
268
269
### Utility Functions Usage
270
271
```python
272
from statemachine.utils import qualname, ensure_iterable, run_async_from_sync
273
274
# Get fully qualified class name
275
name = qualname(OrderMachine)
276
print(name) # "__main__.OrderMachine"
277
278
# Ensure iterable
279
items = ensure_iterable("single_item")
280
print(items) # ("single_item",)
281
282
items = ensure_iterable(["already", "iterable"])
283
print(items) # ["already", "iterable"]
284
285
# Run async from sync context
286
async def async_operation():
287
await asyncio.sleep(0.1)
288
return "completed"
289
290
result = run_async_from_sync(async_operation())
291
print(result) # "completed"
292
```