0
# Signals
1
2
Django signal system integration for handling email delivery events, enabling custom logging, analytics, error tracking, and post-send processing. The package emits signals for both successful and failed email delivery attempts.
3
4
## Capabilities
5
6
### Email Sent Signal
7
8
Django signal emitted after each email send attempt, providing information about success or failure.
9
10
```python { .api }
11
sendgrid_email_sent = django.dispatch.Signal()
12
13
# Signal arguments:
14
# - sender: The SendgridBackend class
15
# - message: The EmailMessage object that was sent
16
# - fail_flag: Boolean indicating if the send failed
17
```
18
19
### Signal Connection
20
21
Connect handlers to the SendGrid email signal for custom processing.
22
23
```python { .api }
24
from sendgrid_backend.signals import sendgrid_email_sent
25
26
@receiver(sendgrid_email_sent)
27
def handle_sendgrid_email(sender, message, fail_flag, **kwargs):
28
"""
29
Handle SendGrid email send events.
30
31
Parameters:
32
- sender: SendgridBackend class (not instance)
33
- message: EmailMessage object with send details
34
- fail_flag: True if send failed, False if successful
35
- **kwargs: Additional signal data
36
"""
37
```
38
39
## Usage Examples
40
41
### Basic Signal Handling
42
43
Simple signal handler for logging email send events:
44
45
```python
46
import logging
47
from django.dispatch import receiver
48
from sendgrid_backend.signals import sendgrid_email_sent
49
50
logger = logging.getLogger(__name__)
51
52
@receiver(sendgrid_email_sent)
53
def log_email_send(sender, message, fail_flag, **kwargs):
54
"""Log all email send attempts."""
55
56
status = "FAILED" if fail_flag else "SUCCESS"
57
recipients = ", ".join(message.to)
58
subject = message.subject
59
60
logger.info(f"Email {status}: '{subject}' to {recipients}")
61
62
if fail_flag:
63
logger.error(f"Email send failed for message: {subject}")
64
else:
65
# Log successful send with message ID if available
66
message_id = message.extra_headers.get('message_id')
67
if message_id:
68
logger.info(f"SendGrid message ID: {message_id}")
69
```
70
71
### Database Logging
72
73
Store email send events in database for analytics and tracking:
74
75
```python
76
from django.db import models
77
from django.dispatch import receiver
78
from django.utils import timezone
79
from sendgrid_backend.signals import sendgrid_email_sent
80
81
class EmailSendLog(models.Model):
82
"""Model to track email send events."""
83
84
subject = models.CharField(max_length=255)
85
recipients = models.TextField() # JSON or comma-separated
86
sender = models.EmailField()
87
status = models.CharField(max_length=20)
88
sendgrid_message_id = models.CharField(max_length=100, blank=True)
89
send_time = models.DateTimeField(default=timezone.now)
90
failure_reason = models.TextField(blank=True)
91
92
# Additional tracking fields
93
template_id = models.CharField(max_length=50, blank=True)
94
categories = models.JSONField(default=list, blank=True)
95
custom_args = models.JSONField(default=dict, blank=True)
96
97
@receiver(sendgrid_email_sent)
98
def log_email_to_database(sender, message, fail_flag, **kwargs):
99
"""Store email send events in database."""
100
101
# Extract message details
102
recipients_list = message.to + message.cc + message.bcc
103
recipients = ",".join(recipients_list)
104
105
# Get SendGrid-specific data
106
template_id = getattr(message, 'template_id', '')
107
categories = getattr(message, 'categories', [])
108
custom_args = getattr(message, 'custom_args', {})
109
110
# Get message ID from headers (available after successful send)
111
message_id = message.extra_headers.get('message_id', '')
112
113
# Create log entry
114
EmailSendLog.objects.create(
115
subject=message.subject,
116
recipients=recipients,
117
sender=message.from_email,
118
status='failed' if fail_flag else 'sent',
119
sendgrid_message_id=message_id,
120
template_id=template_id,
121
categories=categories,
122
custom_args=custom_args,
123
failure_reason='' if not fail_flag else 'Send attempt failed'
124
)
125
```
126
127
### Metrics and Analytics
128
129
Collect metrics on email sending patterns and performance:
130
131
```python
132
from django.core.cache import cache
133
from django.dispatch import receiver
134
from sendgrid_backend.signals import sendgrid_email_sent
135
import json
136
137
@receiver(sendgrid_email_sent)
138
def collect_email_metrics(sender, message, fail_flag, **kwargs):
139
"""Collect email sending metrics for analytics."""
140
141
# Increment counters
142
today = timezone.now().date().isoformat()
143
144
if fail_flag:
145
cache_key = f"email_failures_{today}"
146
cache.get_or_set(cache_key, 0, timeout=86400)
147
cache.incr(cache_key)
148
149
# Track failure reasons if available
150
failure_key = f"email_failure_reasons_{today}"
151
failures = cache.get(failure_key, {})
152
failure_type = "unknown" # Could extract from exception context
153
failures[failure_type] = failures.get(failure_type, 0) + 1
154
cache.set(failure_key, failures, timeout=86400)
155
156
else:
157
cache_key = f"email_successes_{today}"
158
cache.get_or_set(cache_key, 0, timeout=86400)
159
cache.incr(cache_key)
160
161
# Track successful sends by template if applicable
162
template_id = getattr(message, 'template_id', None)
163
if template_id:
164
template_key = f"template_usage_{today}"
165
usage = cache.get(template_key, {})
166
usage[template_id] = usage.get(template_id, 0) + 1
167
cache.set(template_key, usage, timeout=86400)
168
169
# Track recipient domains
170
domains_key = f"recipient_domains_{today}"
171
domains = cache.get(domains_key, {})
172
173
for email in message.to:
174
domain = email.split('@')[-1] if '@' in email else 'unknown'
175
domains[domain] = domains.get(domain, 0) + 1
176
177
cache.set(domains_key, domains, timeout=86400)
178
179
def get_email_metrics(date=None):
180
"""Retrieve email metrics for a specific date."""
181
if date is None:
182
date = timezone.now().date().isoformat()
183
184
return {
185
'successes': cache.get(f"email_successes_{date}", 0),
186
'failures': cache.get(f"email_failures_{date}", 0),
187
'failure_reasons': cache.get(f"email_failure_reasons_{date}", {}),
188
'template_usage': cache.get(f"template_usage_{date}", {}),
189
'recipient_domains': cache.get(f"recipient_domains_{date}", {})
190
}
191
```
192
193
### Error Handling and Alerting
194
195
Set up alerts for email send failures and performance issues:
196
197
```python
198
from django.dispatch import receiver
199
from django.core.mail import mail_administrators
200
from sendgrid_backend.signals import sendgrid_email_sent
201
import logging
202
203
logger = logging.getLogger(__name__)
204
205
@receiver(sendgrid_email_sent)
206
def handle_email_failures(sender, message, fail_flag, **kwargs):
207
"""Handle email send failures with alerting."""
208
209
if not fail_flag:
210
return # Only process failures
211
212
# Log the failure
213
logger.error(f"SendGrid email send failed: {message.subject}")
214
215
# Check failure rate
216
failure_rate = check_recent_failure_rate()
217
218
if failure_rate > 0.1: # More than 10% failure rate
219
send_admin_alert(f"High email failure rate: {failure_rate:.1%}")
220
221
# Alert for specific critical emails
222
is_critical = any(category in getattr(message, 'categories', [])
223
for category in ['critical', 'transactional', 'password-reset'])
224
225
if is_critical:
226
send_admin_alert(f"Critical email failed: {message.subject}")
227
228
# Store failed message for retry
229
store_failed_message(message)
230
231
def check_recent_failure_rate():
232
"""Calculate recent email failure rate."""
233
now = timezone.now()
234
hour_ago = now - timedelta(hours=1)
235
236
recent_logs = EmailSendLog.objects.filter(send_time__gte=hour_ago)
237
total = recent_logs.count()
238
239
if total == 0:
240
return 0
241
242
failures = recent_logs.filter(status='failed').count()
243
return failures / total
244
245
def send_admin_alert(message):
246
"""Send alert to administrators."""
247
mail_administrators(
248
subject='SendGrid Email Alert',
249
message=message,
250
fail_silently=True # Don't fail if admin email fails
251
)
252
253
def store_failed_message(message):
254
"""Store failed message for potential retry."""
255
# Implementation depends on your retry strategy
256
# Could store in database, queue, or cache
257
pass
258
```
259
260
### Integration with External Services
261
262
Send email events to external analytics or monitoring services:
263
264
```python
265
import requests
266
from django.dispatch import receiver
267
from django.conf import settings
268
from sendgrid_backend.signals import sendgrid_email_sent
269
270
@receiver(sendgrid_email_sent)
271
def send_to_analytics(sender, message, fail_flag, **kwargs):
272
"""Send email events to external analytics service."""
273
274
if not hasattr(settings, 'ANALYTICS_WEBHOOK_URL'):
275
return
276
277
# Prepare event data
278
event_data = {
279
'event_type': 'email_send_failed' if fail_flag else 'email_send_success',
280
'timestamp': timezone.now().isoformat(),
281
'email_subject': message.subject,
282
'recipient_count': len(message.to),
283
'template_id': getattr(message, 'template_id', None),
284
'categories': getattr(message, 'categories', []),
285
'custom_args': getattr(message, 'custom_args', {}),
286
}
287
288
# Add SendGrid message ID for successful sends
289
if not fail_flag:
290
event_data['sendgrid_message_id'] = message.extra_headers.get('message_id')
291
292
# Send to analytics service (async recommended for production)
293
try:
294
requests.post(
295
settings.ANALYTICS_WEBHOOK_URL,
296
json=event_data,
297
timeout=5
298
)
299
except requests.RequestException:
300
# Log but don't fail the email send process
301
logger.warning("Failed to send email event to analytics service")
302
303
@receiver(sendgrid_email_sent)
304
def update_user_email_status(sender, message, fail_flag, **kwargs):
305
"""Update user records with email delivery status."""
306
307
# Extract user information from message
308
user_emails = message.to + message.cc + message.bcc
309
310
for email in user_emails:
311
try:
312
# Update user's email status
313
user = User.objects.get(email=email)
314
315
if fail_flag:
316
# Increment failure count
317
profile = user.profile
318
profile.email_failures = profile.email_failures + 1
319
profile.last_email_failure = timezone.now()
320
profile.save()
321
322
# Disable email for high failure rates
323
if profile.email_failures > 5:
324
profile.email_enabled = False
325
profile.save()
326
327
else:
328
# Reset failure count on success
329
profile = user.profile
330
profile.email_failures = 0
331
profile.last_successful_email = timezone.now()
332
profile.save()
333
334
except User.DoesNotExist:
335
# Email not associated with user account
336
continue
337
```
338
339
### Signal Configuration
340
341
Best practices for configuring signal handlers:
342
343
```python
344
# apps.py
345
from django.apps import AppConfig
346
347
class YourAppConfig(AppConfig):
348
name = 'your_app'
349
350
def ready(self):
351
# Import signal handlers to register them
352
import your_app.signals # noqa
353
354
# signals.py - organize all signal handlers in one module
355
from django.dispatch import receiver
356
from sendgrid_backend.signals import sendgrid_email_sent
357
358
# Import all your signal handlers
359
from .handlers.logging import log_email_send
360
from .handlers.metrics import collect_email_metrics
361
from .handlers.alerts import handle_email_failures
362
363
# Signal handlers are automatically registered via @receiver decorator
364
```
365
366
### Testing Signal Handlers
367
368
Test signal handlers to ensure proper functionality:
369
370
```python
371
from django.test import TestCase
372
from django.core.mail import EmailMessage
373
from sendgrid_backend.signals import sendgrid_email_sent
374
375
class SignalHandlerTestCase(TestCase):
376
377
def test_email_send_signal(self):
378
"""Test that signal handlers are called correctly."""
379
380
# Create test message
381
message = EmailMessage(
382
subject='Test Email',
383
body='Test content',
384
from_email='test@example.com',
385
to=['recipient@example.com']
386
)
387
388
# Mock signal handler
389
handler_called = False
390
391
def test_handler(sender, message, fail_flag, **kwargs):
392
nonlocal handler_called
393
handler_called = True
394
self.assertFalse(fail_flag) # Should be successful
395
self.assertEqual(message.subject, 'Test Email')
396
397
# Connect test handler
398
sendgrid_email_sent.connect(test_handler)
399
400
try:
401
# Send signal manually for testing
402
sendgrid_email_sent.send(
403
sender=type(None),
404
message=message,
405
fail_flag=False
406
)
407
408
# Verify handler was called
409
self.assertTrue(handler_called)
410
411
finally:
412
# Disconnect test handler
413
sendgrid_email_sent.disconnect(test_handler)
414
```