Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Django FirstWorkingEmailBackend uses ESP status APIs to find an "up" ESP for sending
# *Untested* implementation of a Django email backend that checks
# several ESP status APIs to see which are working, and uses the first
# available one to send.
#
# This caches the ESP API status check results for one hour (by default).
# You can subscribe to webhook notifications on the ESP status pages
# to force status re-checks after ESPs update their status pages.
#
# See https://github.com/anymail/django-anymail/issues/31 for discussion.
#
# Usage: save this file into your own Django project, and then...
# ... in settings.py:
#
# EMAIL_BACKEND = 'path.to.this.module.FirstWorkingEmailBackend'
#
# FIRST_WORKING_EMAIL_BACKENDS = [
# # Set this to a list of all desired email backends, in order of preference:
# 'anymail.backends.sendgrid.SendGridBackend',
# 'anymail.backends.postmark.PostmarkBackend',
# 'anymail.backends.mailgun.MailgunBackend',
# 'anymail.backends.sparkpost.SparkPostBackend',
# ]
# # Optionally set how long to wait between status checks
# # default is one hour; set to None to check on every send:
# FIRST_WORKING_EMAIL_CACHE_TIMEOUT = 60 * 60 # optional; in seconds
#
# ANYMAIL = {
# # be sure to include settings for *all* backends you've listed
# }
#
# ... in urls.py:
# urlpatterns = [
# ...
# url(r'^esp_status_change/$',
# 'path.to.this.module.invalidate_current_working_email_backend'),
# ...
# ]
#
# And then sign up your 'esp_status_change' url for webhook notifications
# at each ESP's status page.
from django.core.cache import cache
from django.core.exceptions import ImproperlyConfigured
from django.core.mail import get_connection
from django.core.mail.backends.base import BaseEmailBackend
from django.conf import settings
from django.http import HttpResponse
import requests
class FirstWorkingEmailBackend(BaseEmailBackend):
def __init__(self, *args, **kwargs):
super(FirstWorkingEmailBackend, self).__init__(*args, **kwargs)
self._connection = None
def open(self):
if self._connection:
return False
backend_name = find_first_working_backend()
self._connection = get_connection(backend_name, fail_silently=self.fail_silently)
return True
def close(self):
if self._connection:
self._connection.close()
self._connection = None
def send_messages(self, email_messages):
created_connection = self.open()
try:
return self._connection.send_messages(email_messages)
finally:
if created_connection:
self.close()
# How to check whether a backend is "working".
# Loads status api 'url', and looks in response for 'up_value' at 'json_path'.
ESP_STATUS_APIS = {
'anymail.backends.mailgun.MailgunBackend': {
# Use the (undocumented?) statuspage.io summary API.
# Rollup json['status']['indicator'] == 'none' means no problems.
'url': 'http://status.mailgun.com/api/v2/summary.json',
'json_path': 'status.indicator',
'up_value': 'none',
'components_url': 'http://status.mailgun.com/api/v2/components.json',
'components_to_check': [
'Outbound Delivery',
],
'component_up_value': 'operational',
},
'anymail.backends.postmark.PostmarkBackend': {
'url': 'https://status.postmarkapp.com/api/1.0/status',
'json_path': 'status',
'up_value': 'UP',
},
'anymail.backends.sendgrid.SendGridBackend': {
# Alternative approach to statuspage.io summary API.
# json['incidents'] == [] means no problems (empty active incidents list)
'url': 'http://status.sendgrid.com/api/v2/summary.json',
'json_path': 'incidents',
'up_value': [],
'components_url': 'http://status.sendgrid.com/api/v2/components.json',
'components_to_check': [
'Mail Sending',
'API',
],
'component_up_value': 'operational',
},
'anymail.backends.sparkpost.SparkPostBackend': {
# Lighter-weight (and also undocumented?) statuspage.io API.
# Doesn't include components or incidents details.
'url': 'http://status.sparkpost.com/api/v2/status.json',
'json_path': 'status.indicator',
'up_value': 'none',
'components_url': 'http://status.sparkpost.com/api/v2/components.json',
'components_to_check': [
'SMTP',
'APIs',
'SMTP Delivery',
],
'component_up_value': 'operational',
},
}
CACHE_KEY = 'current_working_email_backend'
DEFAULT_CACHE_TIMEOUT = 60*60 # seconds
def invalidate_current_working_email_backend(request):
"""View function to be used as ESP status page webhook"""
cache.delete(CACHE_KEY)
return HttpResponse()
def find_first_working_backend():
"""Returns the first email backend whose status API returns OK, else raises RuntimeError
Uses cached value if available.
"""
backend = cache.get(CACHE_KEY)
if backend is None:
try:
backends = settings.FIRST_WORKING_EMAIL_BACKENDS
cache_timeout = settings.get('FIRST_WORKING_EMAIL_CACHE_TIMEOUT', DEFAULT_CACHE_TIMEOUT)
except AttributeError:
raise ImproperlyConfigured("Set FIRST_WORKING_EMAIL_BACKENDS to a list of possible backends")
for possible_backend in backends:
if is_backend_working(possible_backend):#could be changed to is_component_working
backend = possible_backend
break
if backend is None:
raise RuntimeError("No currently working backend among %s" % ', '.join(backends))
elif cache_timeout:
cache.set(CACHE_KEY, backend)
return backend
def is_backend_working(backend):
try:
status_api = ESP_STATUS_APIS[backend]
except KeyError:
raise ImproperlyConfigured("Don't know how to check ESP %r is working; "
"add it to ESP_STATUS_APIS " % backend)
try:
response = requests.get(status_api['url'])
response.raise_for_status()
json = response.json()
except (requests.RequestException, ValueError):
return False # status API down, error, or non-json response
status = json_path(json, status_api['json_path'])
return status == status_api['up_value']
def is_component_working(backend):
try:
status_api = ESP_STATUS_APIS[backend]
except KeyError:
raise ImproperlyConfigured("Don't know how to check ESP %r is working; "
"add it to ESP_STATUS_APIS " % backend)
try:
response = requests.get(status_api['url'])
response.raise_for_status()
json = response.json()
except (requests.RequestException, ValueError):
return False # status API down, error, or non-json response
for component in json['components']:
if component['name'] in status_api['components_to_check'] and component['status'] != status_api['component_up_value']:
return False
return True #all components are up, component no longer exists, or ESP has changed their statuspage
def json_path(json, path, default=None):
"""Given path='p1.p2', returns json['p1']['p2'], or default if not there"""
# (You could switch this to something like pyjq for more flexibility)
try:
result = json
for segment in path.split('.'):
result = result[segment]
return result
except KeyError:
return default
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.