Django FirstWorkingEmailBackend uses ESP status APIs to find an "up" ESP for sending

first_working_email_backend.py

# *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