0
# Actions and Callbacks
1
2
Lifecycle hooks, action handlers, callback registration, and dependency injection for state machine events including entry/exit actions, transition callbacks, and custom event handlers.
3
4
## Capabilities
5
6
### Callback Registration Methods
7
8
Methods for registering callbacks on transitions and events with support for different callback types and priorities.
9
10
```python { .api }
11
class AddCallbacksMixin:
12
"""Mixin providing callback registration functionality."""
13
14
def add_callback(self, callback, group=None, priority=None, specs=None):
15
"""
16
Add callback to be executed during event processing.
17
18
Parameters:
19
- callback: Callable to execute (function, method, or string reference)
20
- group: CallbackGroup enum value (VALIDATORS, CONDITIONS, ACTIONS)
21
- priority: CallbackPriority enum value (HIGH, NORMAL, LOW)
22
- specs: SpecReference flags for callback resolution
23
"""
24
25
def cond(self, *callbacks):
26
"""Add condition callbacks that must evaluate to True."""
27
28
def unless(self, *callbacks):
29
"""Add condition callbacks that must evaluate to False."""
30
31
def on(self, *callbacks):
32
"""Add action callbacks executed during transition."""
33
34
def before(self, *callbacks):
35
"""Add callbacks executed before transition."""
36
37
def after(self, *callbacks):
38
"""Add callbacks executed after transition."""
39
40
def validators(self, *callbacks):
41
"""Add validator callbacks for transition validation."""
42
```
43
44
### Callback Enums and Constants
45
46
Enumerations and constants for callback organization and execution control.
47
48
```python { .api }
49
class CallbackPriority(IntEnum):
50
"""Priority levels for callback execution order."""
51
HIGH = 10
52
NORMAL = 5
53
LOW = 1
54
55
class CallbackGroup(IntEnum):
56
"""Groups for organizing different types of callbacks."""
57
VALIDATORS = 1
58
CONDITIONS = 2
59
ACTIONS = 3
60
61
class SpecReference(IntFlag):
62
"""Reference types for callback specification resolution."""
63
NAME = 1 # String name references
64
CALLABLE = 2 # Direct callable references
65
PROPERTY = 4 # Property references
66
67
# Constants for common specification combinations
68
SPECS_ALL: SpecReference = NAME | CALLABLE | PROPERTY # All reference types
69
SPECS_SAFE: SpecReference = NAME # Only name references for safety
70
```
71
72
### Convention-Based Callback Methods
73
74
Standard callback method naming conventions that are automatically discovered and registered.
75
76
```python { .api }
77
# Event-based callbacks (replace {event} with actual event name)
78
def before_{event}(self, event: str, source: State, target: State, **kwargs):
79
"""Called before event processing starts."""
80
81
def after_{event}(self, event: str, source: State, target: State, **kwargs):
82
"""Called after event processing completes."""
83
84
def on_{event}(self, event: str, source: State, target: State, **kwargs):
85
"""Called during event transition execution."""
86
87
# State-based callbacks (replace {state} with actual state id)
88
def on_enter_{state}(self, **kwargs):
89
"""Called when entering the specified state."""
90
91
def on_exit_{state}(self, **kwargs):
92
"""Called when exiting the specified state."""
93
94
# Generic lifecycle callbacks
95
def on_enter_state(self, state: State, **kwargs):
96
"""Called when entering any state."""
97
98
def on_exit_state(self, state: State, **kwargs):
99
"""Called when exiting any state."""
100
101
def on_transition(self, event: str, source: State, target: State, **kwargs):
102
"""Called during any transition."""
103
```
104
105
### Built-in Parameters for Dependency Injection
106
107
Parameters automatically available to callback methods through dependency injection.
108
109
```python { .api }
110
# Built-in parameters available to all callbacks:
111
# - event: str - The event identifier
112
# - machine: StateMachine - The state machine instance
113
# - model: object - The external model object (if provided)
114
# - source: State - The source state of transition
115
# - target: State - The target state of transition
116
# - state: State - The current state
117
# - transition: Transition - The transition object
118
# - event_data: TriggerData - Complete event data structure
119
120
def example_callback(self, event: str, source: State, target: State,
121
machine, model, transition, event_data,
122
custom_param: str = "default"):
123
"""Example showing available built-in parameters."""
124
pass
125
```
126
127
## Usage Examples
128
129
### Basic State Entry/Exit Actions
130
131
```python
132
from statemachine import StateMachine, State
133
134
class ConnectionMachine(StateMachine):
135
disconnected = State(initial=True)
136
connecting = State()
137
connected = State()
138
error = State()
139
140
connect = disconnected.to(connecting)
141
success = connecting.to(connected)
142
failure = connecting.to(error)
143
disconnect = connected.to(disconnected)
144
retry = error.to(connecting)
145
146
def on_enter_connecting(self):
147
"""Called when entering connecting state."""
148
print("Attempting to connect...")
149
self.connection_attempts = getattr(self, 'connection_attempts', 0) + 1
150
151
def on_exit_connecting(self):
152
"""Called when leaving connecting state."""
153
print("Connection attempt finished")
154
155
def on_enter_connected(self):
156
"""Called when successfully connected."""
157
print("Connected successfully!")
158
self.connection_attempts = 0
159
160
def on_enter_error(self):
161
"""Called when connection fails."""
162
print(f"Connection failed (attempt {self.connection_attempts})")
163
164
def on_exit_error(self):
165
"""Called when leaving error state."""
166
print("Retrying connection...")
167
168
# Usage
169
conn = ConnectionMachine()
170
conn.send("connect") # Prints: "Attempting to connect..."
171
conn.send("success") # Prints: "Connection attempt finished" then "Connected successfully!"
172
```
173
174
### Event-Based Callbacks with Parameters
175
176
```python
177
class OrderProcessingMachine(StateMachine):
178
created = State(initial=True)
179
validated = State()
180
paid = State()
181
shipped = State()
182
delivered = State(final=True)
183
cancelled = State(final=True)
184
185
validate = created.to(validated)
186
pay = validated.to(paid)
187
ship = paid.to(shipped)
188
deliver = shipped.to(delivered)
189
cancel = (
190
created.to(cancelled)
191
| validated.to(cancelled)
192
| paid.to(cancelled)
193
)
194
195
def before_validate(self, event: str, order_data: dict = None):
196
"""Called before validation starts."""
197
print(f"Starting {event} with order: {order_data.get('id', 'unknown')}")
198
return f"Validation started for order {order_data.get('id')}"
199
200
def after_validate(self, event: str, source: State, target: State,
201
order_data: dict = None):
202
"""Called after validation completes."""
203
print(f"Completed {event}: {source.id} -> {target.id}")
204
if order_data:
205
print(f"Order {order_data['id']} validation successful")
206
207
def before_pay(self, payment_method: str = "card", amount: float = 0.0):
208
"""Called before payment processing."""
209
print(f"Processing payment of ${amount:.2f} via {payment_method}")
210
211
def on_pay(self, payment_method: str = "card", amount: float = 0.0):
212
"""Called during payment processing."""
213
# Simulate payment processing
214
if amount > 0:
215
print(f"Payment of ${amount:.2f} processed successfully")
216
return True
217
return False
218
219
def before_cancel(self, reason: str = "user_request"):
220
"""Called before cancellation."""
221
print(f"Cancelling order. Reason: {reason}")
222
223
# Usage with event parameters
224
order = OrderProcessingMachine()
225
result = order.send("validate", order_data={"id": "ORD-001", "items": ["item1"]})
226
print(f"Result: {result}") # Prints return value from before_validate
227
228
order.send("pay", payment_method="credit_card", amount=99.99)
229
order.send("cancel", reason="inventory_unavailable")
230
```
231
232
### Conditional Callbacks and Validators
233
234
```python
235
class AccountMachine(StateMachine):
236
inactive = State(initial=True)
237
active = State()
238
suspended = State()
239
closed = State(final=True)
240
241
activate = inactive.to(active)
242
suspend = active.to(suspended)
243
reactivate = suspended.to(active)
244
close = (
245
inactive.to(closed)
246
| suspended.to(closed)
247
| active.to(closed)
248
)
249
250
def before_activate(self, user_id: int = None, **kwargs):
251
"""Validator called before activation."""
252
if not user_id:
253
raise ValueError("User ID required for activation")
254
if not self.user_exists(user_id):
255
raise ValueError(f"User {user_id} not found")
256
print(f"Activating account for user {user_id}")
257
258
def on_activate(self, user_id: int = None):
259
"""Action performed during activation."""
260
self.current_user_id = user_id
261
print(f"Account activated for user {user_id}")
262
263
def before_suspend(self, reason: str = "policy_violation"):
264
"""Called before suspension with reason."""
265
print(f"Suspending account. Reason: {reason}")
266
self.suspension_reason = reason
267
268
def before_close(self, **kwargs):
269
"""Validator for account closure."""
270
if hasattr(self, 'current_user_id'):
271
print(f"Closing account for user {self.current_user_id}")
272
else:
273
print("Closing inactive account")
274
275
def user_exists(self, user_id: int) -> bool:
276
"""Simulate user validation."""
277
return user_id > 0 # Simplified validation
278
279
# Usage
280
account = AccountMachine()
281
account.send("activate", user_id=12345)
282
account.send("suspend", reason="suspicious_activity")
283
account.send("close")
284
```
285
286
### Programmatic Callback Registration
287
288
```python
289
class DynamicMachine(StateMachine):
290
state1 = State(initial=True)
291
state2 = State()
292
state3 = State(final=True)
293
294
transition1 = state1.to(state2)
295
transition2 = state2.to(state3)
296
297
def __init__(self, *args, **kwargs):
298
super().__init__(*args, **kwargs)
299
300
# Add callbacks programmatically
301
self.transition1.before(self.log_transition)
302
self.transition1.cond(self.check_condition)
303
self.transition1.on(self.perform_action)
304
305
# Add multiple callbacks with priorities
306
self.transition2.add_callback(
307
self.high_priority_action,
308
group=CallbackGroup.ACTIONS,
309
priority=CallbackPriority.HIGH
310
)
311
self.transition2.add_callback(
312
self.low_priority_action,
313
group=CallbackGroup.ACTIONS,
314
priority=CallbackPriority.LOW
315
)
316
317
def log_transition(self, event: str, source: State, target: State):
318
"""Logging callback."""
319
print(f"Transitioning: {event} ({source.id} -> {target.id})")
320
321
def check_condition(self, allow_transition: bool = True):
322
"""Condition callback."""
323
print(f"Checking condition: {allow_transition}")
324
return allow_transition
325
326
def perform_action(self):
327
"""Action callback."""
328
print("Performing main action")
329
330
def high_priority_action(self):
331
"""High priority action (executed first)."""
332
print("High priority action")
333
334
def low_priority_action(self):
335
"""Low priority action (executed last)."""
336
print("Low priority action")
337
338
# Usage
339
machine = DynamicMachine()
340
machine.send("transition1", allow_transition=True)
341
machine.send("transition2")
342
```
343
344
### Async Callbacks
345
346
```python
347
import asyncio
348
from statemachine import StateMachine, State
349
350
class AsyncWorkflowMachine(StateMachine):
351
pending = State(initial=True)
352
processing = State()
353
completed = State(final=True)
354
failed = State(final=True)
355
356
start = pending.to(processing)
357
complete = processing.to(completed)
358
fail = processing.to(failed)
359
360
async def on_enter_processing(self, task_id: str = None, **kwargs):
361
"""Async entry action."""
362
print(f"Starting async processing for task: {task_id}")
363
try:
364
result = await self.process_task(task_id)
365
print(f"Processing completed: {result}")
366
except Exception as e:
367
print(f"Processing failed: {e}")
368
# Automatically trigger failure transition
369
await self.send_async("fail")
370
371
async def before_complete(self, **kwargs):
372
"""Async before callback."""
373
print("Finalizing task...")
374
await asyncio.sleep(0.5) # Simulate cleanup
375
print("Finalization complete")
376
377
async def process_task(self, task_id: str):
378
"""Simulate async work."""
379
await asyncio.sleep(2)
380
if task_id == "fail_test":
381
raise Exception("Simulated failure")
382
return f"Result for {task_id}"
383
384
# Usage
385
async def main():
386
workflow = AsyncWorkflowMachine()
387
await workflow.send_async("start", task_id="task_001")
388
# Processing will complete automatically or fail
389
if workflow.processing.is_active:
390
await workflow.send_async("complete")
391
392
print(f"Final state: {workflow.current_state.id}")
393
394
asyncio.run(main())
395
```
396
397
### Global State and Transition Callbacks
398
399
```python
400
class AuditedMachine(StateMachine):
401
start = State(initial=True)
402
middle = State()
403
end = State(final=True)
404
405
next = start.to(middle) | middle.to(end)
406
407
def on_enter_state(self, state: State, **kwargs):
408
"""Called when entering any state."""
409
print(f"Entering state: {state.name}")
410
self.log_state_change("enter", state)
411
412
def on_exit_state(self, state: State, **kwargs):
413
"""Called when exiting any state."""
414
print(f"Exiting state: {state.name}")
415
self.log_state_change("exit", state)
416
417
def on_transition(self, event: str, source: State, target: State, **kwargs):
418
"""Called during any transition."""
419
print(f"Transition: {event} ({source.id} -> {target.id})")
420
self.audit_log.append({
421
"event": event,
422
"from": source.id,
423
"to": target.id,
424
"timestamp": self.get_timestamp()
425
})
426
427
def __init__(self, *args, **kwargs):
428
super().__init__(*args, **kwargs)
429
self.audit_log = []
430
self.state_history = []
431
432
def log_state_change(self, action: str, state: State):
433
"""Log state changes."""
434
self.state_history.append(f"{action}:{state.id}")
435
436
def get_timestamp(self):
437
"""Get current timestamp."""
438
import datetime
439
return datetime.datetime.now().isoformat()
440
441
# Usage
442
machine = AuditedMachine()
443
machine.send("next") # start -> middle
444
machine.send("next") # middle -> end
445
446
print("Audit log:", machine.audit_log)
447
print("State history:", machine.state_history)
448
```