Skip to content

Instantly share code, notes, and snippets.

Last active September 6, 2016 20:52
Show Gist options
  • Save medmunds/e6b837bb3b382098d775cb412b889632 to your computer and use it in GitHub Desktop.
Save medmunds/e6b837bb3b382098d775cb412b889632 to your computer and use it in GitHub Desktop.
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 for discussion.
# Usage: save this file into your own Django project, and then...
# ... in
# # 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
# # be sure to include settings for *all* backends you've listed
# }
# ... in
# urlpatterns = [
# ...
# url(r'^esp_status_change/$',
# ''),
# ...
# ]
# 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 = None
def send_messages(self, email_messages):
created_connection =
return self._connection.send_messages(email_messages)
if created_connection:
# How to check whether a backend is "working".
# Loads status api 'url', and looks in response for 'up_value' at 'json_path'.
'anymail.backends.mailgun.MailgunBackend': {
# Use the (undocumented?) summary API.
# Rollup json['status']['indicator'] == 'none' means no problems.
'url': '',
'json_path': 'status.indicator',
'up_value': 'none',
'anymail.backends.postmark.PostmarkBackend': {
'url': '',
'json_path': 'status',
'up_value': 'UP',
'anymail.backends.sendgrid.SendGridBackend': {
# Alternative approach to summary API.
# json['incidents'] == [] means no problems (empty active incidents list)
'url': '',
'json_path': 'incidents',
'up_value': [],
'anymail.backends.sparkpost.SparkPostBackend': {
# Lighter-weight (and also undocumented?) API.
# Doesn't include components or incidents details.
'url': '',
'json_path': 'status.indicator',
'up_value': 'none',
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"""
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:
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):
backend = possible_backend
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):
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)
response = requests.get(status_api['url'])
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 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)
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