0
# Signal System and Events
1

2
Django signals for hooking into state transition lifecycle events, enabling custom logic before and after transitions.
3

4
## Capabilities
5

6
### pre_transition Signal
7

8
Fired before state transition execution, allowing you to implement validation, logging, or preparatory actions.
9

10
```python { .api }
11
# From django_fsm.signals
12
pre_transition = Signal()
13
```
14

15
**Signal Arguments:**
16
- `sender`: Model class
17
- `instance`: Model instance undergoing transition
18
- `name`: Name of the transition method
19
- `field`: FSM field instance
20
- `source`: Current state before transition
21
- `target`: Target state after transition  
22
- `method_args`: Arguments passed to transition method
23
- `method_kwargs`: Keyword arguments passed to transition method
24

25
Usage example:
26

27
```python
28
from django_fsm.signals import pre_transition
29
from django.dispatch import receiver
30
import logging
31

32
@receiver(pre_transition)
33
def log_state_transition(sender, instance, name, source, target, **kwargs):
34
    """Log all state transitions before they occur."""
35
    logging.info(
36
        f"About to transition {sender.__name__} {instance.pk} "
37
        f"from {source} to {target} via {name}"
38
    )
39

40
# For specific models only
41
@receiver(pre_transition, sender=Order)
42
def validate_order_transition(sender, instance, name, source, target, **kwargs):
43
    """Validate order transitions before they happen."""
44
    if name == 'ship' and not instance.payment_confirmed:
45
        raise ValueError("Cannot ship unconfirmed payment")
46
```
47

48
### post_transition Signal
49

50
Fired after successful state transition, enabling cleanup, notifications, or cascade operations.
51

52
```python { .api }
53
# From django_fsm.signals  
54
post_transition = Signal()
55
```
56

57
**Signal Arguments:**
58
- `sender`: Model class
59
- `instance`: Model instance that transitioned
60
- `name`: Name of the transition method
61
- `field`: FSM field instance
62
- `source`: Previous state before transition
63
- `target`: New state after transition
64
- `method_args`: Arguments passed to transition method
65
- `method_kwargs`: Keyword arguments passed to transition method
66
- `exception`: Exception instance (only present if transition failed with on_error state)
67

68
Usage example:
69

70
```python
71
from django_fsm.signals import post_transition
72
from django.dispatch import receiver
73

74
@receiver(post_transition)
75
def handle_state_change(sender, instance, name, source, target, **kwargs):
76
    """Handle any state change across all models."""
77
    print(f"State changed: {sender.__name__} {instance.pk} -> {target}")
78

79
@receiver(post_transition, sender=Order)
80
def order_state_changed(sender, instance, name, source, target, **kwargs):
81
    """Handle order-specific state changes."""
82
    if target == 'shipped':
83
        # Send shipping notification
84
        send_shipping_notification(instance)
85
    elif target == 'cancelled':
86
        # Process refund
87
        process_refund(instance)
88
```
89

90
## Signal Implementation Patterns
91

92
### Audit Trail Implementation
93

94
Create comprehensive audit logs using signals:
95

96
```python
97
from django_fsm.signals import pre_transition, post_transition
98
from django.contrib.auth import get_user_model
99
import json
100

101
class StateTransitionAudit(models.Model):
102
    content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
103
    object_id = models.PositiveIntegerField()
104
    content_object = GenericForeignKey('content_type', 'object_id')
105
    
106
    transition_name = models.CharField(max_length=100)
107
    source_state = models.CharField(max_length=100)
108
    target_state = models.CharField(max_length=100)
109
    user = models.ForeignKey(get_user_model(), null=True, on_delete=models.SET_NULL)
110
    timestamp = models.DateTimeField(auto_now_add=True)
111
    metadata = models.JSONField(default=dict)
112

113
@receiver(post_transition)
114
def create_audit_record(sender, instance, name, source, target, **kwargs):
115
    """Create audit record for every state transition."""
116
    StateTransitionAudit.objects.create(
117
        content_object=instance,
118
        transition_name=name,
119
        source_state=source,
120
        target_state=target,
121
        user=getattr(instance, '_current_user', None),
122
        metadata={
123
            'method_args': kwargs.get('method_args', []),
124
            'method_kwargs': kwargs.get('method_kwargs', {}),
125
            'has_exception': 'exception' in kwargs
126
        }
127
    )
128
```
129

130
### Notification System
131

132
Implement notifications based on state changes:
133

134
```python
135
from django_fsm.signals import post_transition
136
from django.core.mail import send_mail
137

138
@receiver(post_transition, sender=Order)
139
def send_order_notifications(sender, instance, name, source, target, **kwargs):
140
    """Send notifications based on order state changes."""
141
    
142
    notification_map = {
143
        'confirmed': {
144
            'subject': 'Order Confirmed',
145
            'template': 'order_confirmed.html',
146
            'recipients': [instance.customer.email]
147
        },
148
        'shipped': {
149
            'subject': 'Order Shipped', 
150
            'template': 'order_shipped.html',
151
            'recipients': [instance.customer.email]
152
        },
153
        'delivered': {
154
            'subject': 'Order Delivered',
155
            'template': 'order_delivered.html', 
156
            'recipients': [instance.customer.email, instance.sales_rep.email]
157
        }
158
    }
159
    
160
    if target in notification_map:
161
        config = notification_map[target]
162
        send_notification(instance, config)
163

164
def send_notification(order, config):
165
    """Send email notification using template."""
166
    from django.template.loader import render_to_string
167
    
168
    message = render_to_string(config['template'], {'order': order})
169
    send_mail(
170
        subject=config['subject'],
171
        message=message,
172
        from_email='noreply@example.com',
173
        recipient_list=config['recipients']
174
    )
175
```
176

177
### Cascade State Changes
178

179
Trigger related object state changes:
180

181
```python
182
@receiver(post_transition, sender=Order)
183
def cascade_order_state_changes(sender, instance, name, source, target, **kwargs):
184
    """Cascade state changes to related objects."""
185
    
186
    if target == 'cancelled':
187
        # Cancel all order items
188
        for item in instance.orderitem_set.all():
189
            if hasattr(item, 'cancel') and can_proceed(item.cancel):
190
                item.cancel()
191
                item.save()
192
        
193
        # Release reserved inventory
194
        for item in instance.orderitem_set.all():
195
            item.product.release_inventory(item.quantity)
196
    
197
    elif target == 'shipped':
198
        # Update inventory for shipped items
199
        for item in instance.orderitem_set.all():
200
            item.product.reduce_inventory(item.quantity)
201
```
202

203
### Error Handling in Signals
204

205
Handle exceptions in signal receivers:
206

207
```python
208
from django_fsm.signals import post_transition
209
import logging
210

211
logger = logging.getLogger('fsm_signals')
212

213
@receiver(post_transition)
214
def safe_notification_handler(sender, instance, name, source, target, **kwargs):
215
    """Send notifications with error handling."""
216
    try:
217
        if target == 'published':
218
            send_publication_notifications(instance)
219
    except Exception as e:
220
        logger.error(
221
            f"Failed to send notification for {sender.__name__} {instance.pk}: {e}",
222
            exc_info=True
223
        )
224
        # Don't re-raise - notification failure shouldn't break the transition
225
```
226

227
### Conditional Signal Processing
228

229
Process signals based on specific conditions:
230

231
```python
232
@receiver(post_transition)
233
def conditional_processing(sender, instance, name, source, target, **kwargs):
234
    """Process signals based on conditions."""
235
    
236
    # Only process during business hours
237
    from datetime import datetime
238
    current_hour = datetime.now().hour
239
    if not (9 <= current_hour <= 17):
240
        return
241
    
242
    # Only process specific transitions
243
    if name not in ['publish', 'approve', 'reject']:
244
        return
245
    
246
    # Only for specific models
247
    if sender not in [Article, Document, BlogPost]:
248
        return
249
    
250
    # Process the signal
251
    handle_business_hours_transition(instance, name, target)
252
```
253

254
### Signal-Based Metrics
255

256
Collect metrics using signals:
257

258
```python
259
from django.core.cache import cache
260
from django_fsm.signals import post_transition
261

262
@receiver(post_transition)
263
def collect_transition_metrics(sender, instance, name, source, target, **kwargs):
264
    """Collect metrics about state transitions."""
265
    
266
    # Count transitions by type
267
    cache_key = f"transition_count:{sender.__name__}:{name}"
268
    cache.set(cache_key, cache.get(cache_key, 0) + 1, timeout=3600)
269
    
270
    # Track state distribution
271
    state_key = f"state_count:{sender.__name__}:{target}"
272
    cache.set(state_key, cache.get(state_key, 0) + 1, timeout=3600)
273
    
274
    # Track transition timing
275
    timing_key = f"transition_time:{sender.__name__}:{name}"
276
    transition_time = timezone.now()
277
    cache.set(timing_key, transition_time.isoformat(), timeout=3600)
278
```
279

280
### Testing Signal Handlers
281

282
Test signal-based functionality:
283

284
```python
285
from django.test import TestCase
286
from django_fsm.signals import post_transition
287
from unittest.mock import patch, Mock
288

289
class SignalTests(TestCase):
290
    def test_notification_sent_on_state_change(self):
291
        """Test that notifications are sent when order is shipped."""
292
        
293
        with patch('myapp.signals.send_mail') as mock_send_mail:
294
            order = Order.objects.create(state='confirmed')
295
            order.ship()
296
            order.save()
297
            
298
            # Verify notification was sent
299
            mock_send_mail.assert_called_once()
300
            args, kwargs = mock_send_mail.call_args
301
            self.assertIn('Order Shipped', args[0])  # subject
302
            self.assertIn(order.customer.email, kwargs['recipient_list'])
303
    
304
    def test_signal_handler_exception_handling(self):
305
        """Test that signal handler exceptions don't break transitions."""
306
        
307
        with patch('myapp.signals.send_publication_notifications') as mock_notify:
308
            mock_notify.side_effect = Exception("Notification failed")
309
            
310
            article = Article.objects.create(state='draft')
311
            article.publish()
312
            article.save()
313
            
314
            # Transition should succeed despite notification failure
315
            self.assertEqual(article.state, 'published')
316
```
317

318
### Performance Considerations
319

320
Optimize signal handlers for performance:
321

322
```python
323
from django_fsm.signals import post_transition
324
from django.core.cache import cache
325

326
@receiver(post_transition)
327
def optimized_signal_handler(sender, instance, name, source, target, **kwargs):
328
    """Optimized signal handler with caching and batching."""
329
    
330
    # Use caching to avoid repeated database queries
331
    cache_key = f"config:{sender.__name__}:{name}"
332
    config = cache.get(cache_key)
333
    if config is None:
334
        config = load_transition_config(sender, name)
335
        cache.set(cache_key, config, timeout=300)
336
    
337
    # Batch operations when possible
338
    if target == 'processed':
339
        # Collect IDs for batch processing
340
        batch_key = f"batch_process:{sender.__name__}"
341
        ids = cache.get(batch_key, [])
342
        ids.append(instance.pk)
343
        cache.set(batch_key, ids, timeout=60)
344
        
345
        # Process in batches of 10
346
        if len(ids) >= 10:
347
            process_batch(sender, ids)
348
            cache.delete(batch_key)
349
```