0
# Exceptions and Error Handling
1
2
Error scenarios, exception types, and error handling patterns including transition validation errors, invalid state values, and configuration errors.
3
4
## Capabilities
5
6
### Exception Hierarchy
7
8
Comprehensive exception hierarchy for different types of state machine errors.
9
10
```python { .api }
11
class StateMachineError(Exception):
12
"""
13
Base exception for all state machine errors.
14
15
All exceptions raised by the state machine library inherit from this class,
16
making it easy to catch any state machine-related error.
17
"""
18
19
class InvalidDefinition(StateMachineError):
20
"""
21
Raised when the state machine has a definition error.
22
23
This includes errors in state machine class definition, invalid state
24
configurations, or incorrect transition specifications.
25
"""
26
27
class InvalidStateValue(InvalidDefinition):
28
"""
29
Raised when the current model state value is not mapped to a state definition.
30
31
Occurs when external model has a state value that doesn't correspond
32
to any defined State object in the state machine.
33
34
Attributes:
35
- value: The invalid state value that was encountered
36
"""
37
def __init__(self, value, msg=None): ...
38
39
@property
40
def value(self):
41
"""Get the invalid state value."""
42
43
class AttrNotFound(InvalidDefinition):
44
"""
45
Raised when there's no method or property with the given name.
46
47
Occurs when callback references point to non-existent methods or
48
when accessing undefined state machine attributes.
49
"""
50
51
class TransitionNotAllowed(StateMachineError):
52
"""
53
Raised when there's no transition that can run from the current state.
54
55
This is the most common runtime exception, occurring when trying to
56
trigger an event that has no valid transition from the current state.
57
58
Attributes:
59
- event: The Event object that couldn't be processed
60
- state: The State object where the transition was attempted
61
"""
62
def __init__(self, event: Event, state: State): ...
63
64
@property
65
def event(self) -> Event:
66
"""Get the event that couldn't be processed."""
67
68
@property
69
def state(self) -> State:
70
"""Get the state where transition was attempted."""
71
```
72
73
## Usage Examples
74
75
### Handling TransitionNotAllowed
76
77
```python
78
from statemachine import StateMachine, State
79
from statemachine.exceptions import TransitionNotAllowed
80
81
class LightSwitch(StateMachine):
82
off = State(initial=True)
83
on = State()
84
85
turn_on = off.to(on)
86
turn_off = on.to(off)
87
88
# Basic exception handling
89
switch = LightSwitch()
90
91
try:
92
switch.send("turn_off") # Can't turn off when already off
93
except TransitionNotAllowed as e:
94
print(f"Cannot {e.event.name} when in {e.state.name}")
95
print(f"Current state: {switch.current_state.id}")
96
# Output: Cannot turn_off when in Off
97
# Current state: off
98
99
# Checking allowed events before sending
100
if "turn_on" in [event.id for event in switch.allowed_events]:
101
switch.send("turn_on")
102
else:
103
print("Turn on event not allowed")
104
105
# Alternative: Check if event is allowed
106
if switch.is_allowed("turn_on"):
107
switch.send("turn_on")
108
```
109
110
### Graceful Error Handling with Configuration
111
112
```python
113
class RobustMachine(StateMachine):
114
state1 = State(initial=True)
115
state2 = State()
116
state3 = State(final=True)
117
118
advance = state1.to(state2) | state2.to(state3)
119
120
def __init__(self, *args, **kwargs):
121
# Allow events without transitions (won't raise exceptions)
122
super().__init__(*args, allow_event_without_transition=True, **kwargs)
123
124
def handle_unknown_event(self, event: str):
125
"""Custom handler for events without transitions."""
126
print(f"Event '{event}' ignored - no valid transition")
127
128
# Usage
129
machine = RobustMachine()
130
machine.send("unknown_event") # Won't raise exception
131
machine.send("advance") # Normal transition
132
machine.send("unknown_event") # Still won't raise exception
133
```
134
135
### Handling InvalidStateValue with External Models
136
137
```python
138
from statemachine.exceptions import InvalidStateValue
139
140
class Document:
141
def __init__(self):
142
self.status = "invalid_status" # This will cause an error
143
144
class DocumentMachine(StateMachine):
145
draft = State(initial=True, value="draft_status")
146
review = State(value="review_status")
147
published = State(value="published_status", final=True)
148
149
submit = draft.to(review)
150
publish = review.to(published)
151
152
# Handle invalid external state
153
doc = Document()
154
155
try:
156
machine = DocumentMachine(doc, state_field="status")
157
except InvalidStateValue as e:
158
print(f"Invalid state value: {e.value}")
159
print("Available states:", [s.value for s in DocumentMachine._states()])
160
161
# Fix the model and retry
162
doc.status = "draft_status"
163
machine = DocumentMachine(doc, state_field="status")
164
print(f"Current state: {machine.current_state.id}")
165
```
166
167
### Custom Error Handling in Callbacks
168
169
```python
170
class ValidatedMachine(StateMachine):
171
idle = State(initial=True)
172
processing = State()
173
completed = State(final=True)
174
error = State(final=True)
175
176
start = idle.to(processing)
177
complete = processing.to(completed)
178
fail = processing.to(error)
179
180
def before_start(self, data: dict = None):
181
"""Validator that may raise exceptions."""
182
if not data:
183
raise ValueError("Data is required to start processing")
184
185
if not isinstance(data, dict):
186
raise TypeError("Data must be a dictionary")
187
188
required_fields = ["id", "type", "content"]
189
missing = [field for field in required_fields if field not in data]
190
if missing:
191
raise ValueError(f"Missing required fields: {missing}")
192
193
print("Validation passed")
194
195
def on_start(self, data: dict = None):
196
"""Action that may fail."""
197
try:
198
self.process_data(data)
199
except Exception as e:
200
print(f"Processing failed: {e}")
201
# Trigger failure transition
202
self.send("fail")
203
raise # Re-raise to caller
204
205
def process_data(self, data: dict):
206
"""Simulate data processing that may fail."""
207
if data.get("type") == "error_test":
208
raise RuntimeError("Simulated processing error")
209
print(f"Successfully processed data: {data['id']}")
210
211
# Usage with error handling
212
machine = ValidatedMachine()
213
214
# Test validation errors
215
try:
216
machine.send("start") # Missing data
217
except ValueError as e:
218
print(f"Validation error: {e}")
219
220
try:
221
machine.send("start", data="not_a_dict") # Wrong type
222
except TypeError as e:
223
print(f"Type error: {e}")
224
225
try:
226
machine.send("start", data={"id": "123"}) # Missing fields
227
except ValueError as e:
228
print(f"Validation error: {e}")
229
230
# Test processing error
231
try:
232
machine.send("start", data={"id": "test", "type": "error_test", "content": "test"})
233
except RuntimeError as e:
234
print(f"Processing error: {e}")
235
print(f"Machine state after error: {machine.current_state.id}")
236
237
# Successful processing
238
try:
239
machine = ValidatedMachine() # Reset machine
240
machine.send("start", data={"id": "test", "type": "normal", "content": "test"})
241
machine.send("complete")
242
print(f"Final state: {machine.current_state.id}")
243
except Exception as e:
244
print(f"Unexpected error: {e}")
245
```
246
247
### Error Recovery Patterns
248
249
```python
250
class RecoveryMachine(StateMachine):
251
normal = State(initial=True)
252
error = State()
253
recovery = State()
254
failed = State(final=True)
255
256
trigger_error = normal.to(error)
257
attempt_recovery = error.to(recovery)
258
recover = recovery.to(normal)
259
give_up = error.to(failed) | recovery.to(failed)
260
261
def __init__(self, *args, **kwargs):
262
super().__init__(*args, **kwargs)
263
self.error_count = 0
264
self.max_retries = 3
265
266
def on_enter_error(self, error_info: str = "Unknown error"):
267
"""Handle entering error state."""
268
self.error_count += 1
269
print(f"Error #{self.error_count}: {error_info}")
270
271
if self.error_count <= self.max_retries:
272
print("Attempting recovery...")
273
self.send("attempt_recovery")
274
else:
275
print("Max retries exceeded, giving up")
276
self.send("give_up")
277
278
def on_enter_recovery(self):
279
"""Attempt recovery."""
280
import random
281
if random.choice([True, False]): # 50% success rate
282
print("Recovery successful!")
283
self.error_count = 0 # Reset error count
284
self.send("recover")
285
else:
286
print("Recovery failed")
287
self.send("give_up")
288
289
def on_enter_failed(self):
290
"""Handle permanent failure."""
291
print("System has failed permanently")
292
293
# Usage
294
machine = RecoveryMachine()
295
296
# Simulate multiple errors
297
for i in range(5):
298
try:
299
if machine.current_state.id == "normal":
300
machine.send("trigger_error", error_info=f"Error scenario {i+1}")
301
# Machine handles recovery automatically
302
if machine.current_state.id in ["failed"]:
303
break
304
except Exception as e:
305
print(f"Unexpected error: {e}")
306
break
307
308
print(f"Final state: {machine.current_state.id}")
309
```
310
311
### Debugging and Error Logging
312
313
```python
314
import logging
315
from statemachine.exceptions import StateMachineError
316
317
# Configure logging
318
logging.basicConfig(level=logging.INFO)
319
logger = logging.getLogger(__name__)
320
321
class LoggingMachine(StateMachine):
322
start = State(initial=True)
323
middle = State()
324
end = State(final=True)
325
error = State(final=True)
326
327
proceed = start.to(middle) | middle.to(end)
328
fail = start.to(error) | middle.to(error)
329
330
def on_transition(self, event: str, source: State, target: State, **kwargs):
331
"""Log all transitions."""
332
logger.info(f"Transition: {event} ({source.id} -> {target.id})")
333
334
def on_enter_state(self, state: State, **kwargs):
335
"""Log state entries."""
336
logger.info(f"Entering state: {state.name}")
337
338
def send(self, event: str, *args, **kwargs):
339
"""Override send to add error logging."""
340
try:
341
logger.debug(f"Sending event: {event} with args={args}, kwargs={kwargs}")
342
result = super().send(event, *args, **kwargs)
343
logger.debug(f"Event {event} processed successfully")
344
return result
345
except StateMachineError as e:
346
logger.error(f"State machine error in event {event}: {e}")
347
logger.error(f"Current state: {self.current_state.id}")
348
logger.error(f"Allowed events: {[ev.id for ev in self.allowed_events]}")
349
raise
350
except Exception as e:
351
logger.error(f"Unexpected error in event {event}: {e}")
352
# Try to transition to error state if possible
353
if self.is_allowed("fail"):
354
logger.info("Attempting automatic error state transition")
355
super().send("fail")
356
raise
357
358
# Usage with logging
359
machine = LoggingMachine()
360
machine.send("proceed")
361
try:
362
machine.send("invalid_event")
363
except TransitionNotAllowed:
364
logger.warning("Handled transition not allowed error")
365
366
machine.send("fail") # Transition to error state
367
```