0
# Dynamic State Classes
1
2
Classes for dynamic state resolution allowing transition targets to be determined at runtime based on method return values or custom functions.
3
4
## Capabilities
5
6
### RETURN_VALUE
7
8
Uses the method's return value as the target state, enabling transitions where the destination state is determined by business logic at runtime.
9
10
```python { .api }
11
class RETURN_VALUE(State):
12
def __init__(self, *allowed_states):
13
"""
14
Dynamic state that uses method return value as target state.
15
16
Parameters:
17
- *allowed_states: Optional tuple of allowed return values.
18
If provided, return value must be in this list.
19
"""
20
21
def get_state(self, model, transition, result, args=[], kwargs={}):
22
"""
23
Get target state from method return value.
24
25
Parameters:
26
- model: Model instance
27
- transition: Transition object
28
- result: Return value from transition method
29
- args: Method arguments
30
- kwargs: Method keyword arguments
31
32
Returns:
33
str: Target state
34
35
Raises:
36
InvalidResultState: If result not in allowed_states
37
"""
38
```
39
40
Usage example:
41
42
```python
43
from django_fsm import FSMField, transition, RETURN_VALUE
44
45
class Task(models.Model):
46
state = FSMField(default='new')
47
priority = models.IntegerField(default=1)
48
49
@transition(field=state, source='new', target=RETURN_VALUE())
50
def categorize(self):
51
"""Determine state based on priority."""
52
if self.priority >= 8:
53
return 'urgent'
54
elif self.priority >= 5:
55
return 'normal'
56
else:
57
return 'low'
58
59
# Usage
60
task = Task.objects.create(priority=9)
61
task.categorize()
62
print(task.state) # 'urgent'
63
```
64
65
### RETURN_VALUE with Allowed States
66
67
Restrict the possible return values to a predefined set:
68
69
```python
70
class Document(models.Model):
71
state = FSMField(default='draft')
72
content_score = models.FloatField(default=0.0)
73
74
@transition(
75
field=state,
76
source='review',
77
target=RETURN_VALUE('approved', 'rejected', 'needs_revision')
78
)
79
def review_content(self):
80
"""Review content and return appropriate state."""
81
if self.content_score >= 0.8:
82
return 'approved'
83
elif self.content_score >= 0.5:
84
return 'needs_revision'
85
else:
86
return 'rejected'
87
88
# This will raise InvalidResultState if method returns invalid state
89
try:
90
doc = Document.objects.create(content_score=0.9)
91
doc.review_content() # Returns 'approved' - valid
92
except InvalidResultState as e:
93
print(f"Invalid state returned: {e}")
94
```
95
96
### GET_STATE
97
98
Uses a custom function to determine the target state, providing maximum flexibility for complex state resolution logic.
99
100
```python { .api }
101
class GET_STATE(State):
102
def __init__(self, func, states=None):
103
"""
104
Dynamic state that uses custom function to determine target state.
105
106
Parameters:
107
- func: Function that takes (model, *args, **kwargs) and returns state
108
- states: Optional tuple of allowed states for validation
109
"""
110
111
def get_state(self, model, transition, result, args=[], kwargs={}):
112
"""
113
Get target state using custom function.
114
115
Parameters:
116
- model: Model instance
117
- transition: Transition object
118
- result: Return value from transition method (ignored)
119
- args: Method arguments
120
- kwargs: Method keyword arguments
121
122
Returns:
123
str: Target state from custom function
124
125
Raises:
126
InvalidResultState: If result not in allowed states
127
"""
128
```
129
130
Usage examples:
131
132
```python
133
from django_fsm import GET_STATE
134
135
def determine_approval_state(order, *args, **kwargs):
136
"""Custom function to determine approval state."""
137
if order.amount > 10000:
138
return 'executive_approval'
139
elif order.amount > 1000:
140
return 'manager_approval'
141
else:
142
return 'auto_approved'
143
144
class Order(models.Model):
145
state = FSMField(default='pending')
146
amount = models.DecimalField(max_digits=10, decimal_places=2)
147
148
@transition(
149
field=state,
150
source='pending',
151
target=GET_STATE(determine_approval_state)
152
)
153
def submit_for_approval(self):
154
"""Submit order for appropriate approval level."""
155
# Business logic here
156
self.submitted_at = timezone.now()
157
158
# Usage
159
order = Order.objects.create(amount=Decimal('15000.00'))
160
order.submit_for_approval()
161
print(order.state) # 'executive_approval'
162
```
163
164
### GET_STATE with Validation
165
166
Restrict possible states using the states parameter:
167
168
```python
169
def calculate_risk_level(loan, *args, **kwargs):
170
"""Calculate risk-based state."""
171
risk_score = loan.calculate_risk_score()
172
if risk_score > 0.8:
173
return 'high_risk'
174
elif risk_score > 0.5:
175
return 'medium_risk'
176
else:
177
return 'low_risk'
178
179
class LoanApplication(models.Model):
180
state = FSMField(default='submitted')
181
credit_score = models.IntegerField()
182
income = models.DecimalField(max_digits=10, decimal_places=2)
183
184
@transition(
185
field=state,
186
source='submitted',
187
target=GET_STATE(
188
calculate_risk_level,
189
states=('low_risk', 'medium_risk', 'high_risk')
190
)
191
)
192
def assess_risk(self):
193
pass
194
195
def calculate_risk_score(self):
196
# Complex risk calculation logic
197
if self.credit_score < 600:
198
return 0.9
199
elif self.credit_score < 700:
200
return 0.6
201
else:
202
return 0.3
203
```
204
205
## Advanced Dynamic State Patterns
206
207
### Context-Aware State Resolution
208
209
Use method arguments and context to determine states:
210
211
```python
212
def determine_priority_state(task, urgency_level, department, *args, **kwargs):
213
"""Determine state based on method arguments and model data."""
214
if urgency_level == 'critical':
215
return 'immediate'
216
elif department == 'security' and task.security_sensitive:
217
return 'security_review'
218
elif task.estimated_hours > 40:
219
return 'project_planning'
220
else:
221
return 'standard_queue'
222
223
class Task(models.Model):
224
state = FSMField(default='created')
225
estimated_hours = models.IntegerField(default=1)
226
security_sensitive = models.BooleanField(default=False)
227
228
@transition(
229
field=state,
230
source='created',
231
target=GET_STATE(determine_priority_state)
232
)
233
def prioritize(self, urgency_level, department):
234
"""Prioritize task based on urgency and department."""
235
self.prioritized_at = timezone.now()
236
237
# Usage with arguments
238
task = Task.objects.create(estimated_hours=50)
239
task.prioritize('normal', 'security')
240
print(task.state) # 'project_planning' (due to estimated_hours > 40)
241
```
242
243
### Multi-Factor State Resolution
244
245
Combine multiple factors for complex state determination:
246
247
```python
248
def complex_approval_state(expense, *args, **kwargs):
249
"""Multi-factor state determination."""
250
factors = {
251
'amount': expense.amount,
252
'category': expense.category,
253
'requestor_level': expense.requestor.level,
254
'department_budget': expense.department.remaining_budget
255
}
256
257
# Executive approval needed
258
if factors['amount'] > 50000:
259
return 'executive_approval'
260
261
# Department head approval
262
elif factors['amount'] > 10000 or factors['category'] == 'travel':
263
return 'department_head_approval'
264
265
# Manager approval for mid-level amounts
266
elif factors['amount'] > 1000:
267
return 'manager_approval'
268
269
# Auto-approve for senior staff small expenses
270
elif factors['requestor_level'] >= 3 and factors['amount'] <= 500:
271
return 'auto_approved'
272
273
# Default to manager approval
274
else:
275
return 'manager_approval'
276
277
class ExpenseRequest(models.Model):
278
state = FSMField(default='submitted')
279
amount = models.DecimalField(max_digits=10, decimal_places=2)
280
category = models.CharField(max_length=50)
281
requestor = models.ForeignKey(User, on_delete=models.CASCADE)
282
department = models.ForeignKey(Department, on_delete=models.CASCADE)
283
284
@transition(
285
field=state,
286
source='submitted',
287
target=GET_STATE(complex_approval_state)
288
)
289
def route_approval(self):
290
pass
291
```
292
293
### Time-Based State Resolution
294
295
Determine states based on time factors:
296
297
```python
298
from datetime import datetime, timedelta
299
300
def time_based_expiry_state(subscription, *args, **kwargs):
301
"""Determine state based on subscription timing."""
302
now = timezone.now()
303
expires_at = subscription.expires_at
304
305
if expires_at <= now:
306
return 'expired'
307
elif expires_at <= now + timedelta(days=7):
308
return 'expiring_soon'
309
elif expires_at <= now + timedelta(days=30):
310
return 'renewal_period'
311
else:
312
return 'active'
313
314
class Subscription(models.Model):
315
state = FSMField(default='pending')
316
expires_at = models.DateTimeField()
317
318
@transition(
319
field=state,
320
source='pending',
321
target=GET_STATE(time_based_expiry_state)
322
)
323
def activate(self):
324
# Set expiration date
325
self.expires_at = timezone.now() + timedelta(days=365)
326
```
327
328
### External API State Resolution
329
330
Determine states based on external service responses:
331
332
```python
333
def payment_processor_state(payment, *args, **kwargs):
334
"""Determine state based on external payment processor."""
335
try:
336
response = payment_gateway.check_status(payment.transaction_id)
337
338
status_map = {
339
'completed': 'paid',
340
'pending': 'processing',
341
'failed': 'failed',
342
'refunded': 'refunded',
343
'cancelled': 'cancelled'
344
}
345
346
return status_map.get(response.status, 'unknown')
347
348
except PaymentGatewayError:
349
return 'gateway_error'
350
351
class Payment(models.Model):
352
state = FSMField(default='initialized')
353
transaction_id = models.CharField(max_length=100)
354
355
@transition(
356
field=state,
357
source='initialized',
358
target=GET_STATE(
359
payment_processor_state,
360
states=('paid', 'processing', 'failed', 'refunded', 'cancelled', 'gateway_error', 'unknown')
361
)
362
)
363
def sync_with_gateway(self):
364
pass
365
```
366
367
## Error Handling with Dynamic States
368
369
### Invalid State Handling
370
371
Handle cases where dynamic state resolution fails:
372
373
```python
374
from django_fsm import InvalidResultState
375
376
def risky_state_calculation(model, *args, **kwargs):
377
"""State calculation that might return invalid states."""
378
external_status = get_external_status(model.external_id)
379
# This might return values not in allowed_states
380
return external_status
381
382
class ExternalSync(models.Model):
383
state = FSMField(default='syncing')
384
external_id = models.CharField(max_length=100)
385
386
@transition(
387
field=state,
388
source='syncing',
389
target=GET_STATE(
390
risky_state_calculation,
391
states=('completed', 'failed', 'partial')
392
)
393
)
394
def sync_status(self):
395
try:
396
# Normal transition logic
397
pass
398
except InvalidResultState as e:
399
# Handle invalid state by setting default
400
self.state = 'failed'
401
self.error_message = f"Invalid external status: {e}"
402
raise
403
```
404
405
### Fallback State Logic
406
407
Implement fallback logic when dynamic resolution fails:
408
409
```python
410
def safe_state_calculation(model, *args, **kwargs):
411
"""State calculation with built-in fallback."""
412
try:
413
# Primary state calculation
414
calculated_state = complex_calculation(model)
415
416
# Validate against allowed states
417
allowed = ('approved', 'rejected', 'pending')
418
if calculated_state in allowed:
419
return calculated_state
420
else:
421
# Fallback to safe default
422
return 'pending'
423
424
except Exception:
425
# Error fallback
426
return 'error'
427
```
428
429
### Testing Dynamic States
430
431
Test dynamic state behavior thoroughly:
432
433
```python
434
from django.test import TestCase
435
from django_fsm import InvalidResultState
436
437
class DynamicStateTests(TestCase):
438
def test_return_value_state_resolution(self):
439
"""Test RETURN_VALUE state resolution."""
440
task = Task.objects.create(priority=8)
441
task.categorize()
442
self.assertEqual(task.state, 'urgent')
443
444
def test_get_state_function_resolution(self):
445
"""Test GET_STATE with custom function."""
446
order = Order.objects.create(amount=Decimal('15000'))
447
order.submit_for_approval()
448
self.assertEqual(order.state, 'executive_approval')
449
450
def test_invalid_state_raises_exception(self):
451
"""Test that invalid states raise InvalidResultState."""
452
with patch('myapp.models.risky_state_calculation') as mock_calc:
453
mock_calc.return_value = 'invalid_state'
454
455
sync = ExternalSync.objects.create()
456
with self.assertRaises(InvalidResultState):
457
sync.sync_status()
458
```