Replication tester for postgresql, sends email if replication is broken

test_replication.py

#!/usr/bin/env python
#coding: utf-8

import settings
import subprocess


def get_email_logger(addrs, subject, email_settings):
    ''' Returns logger that send email '''

    import logging
    import logging.handlers

    assert isinstance(email_settings, dict)
    assert 'host' in email_settings, "email settings, host needed"
    assert 'user' in email_settings, "email settings, user needed"
    assert 'password' in email_settings, "email settings, password needed"

    logger = logging.getLogger("email_logging")
    logger.setLevel(logging.DEBUG)

    handler_options = {
        'mailhost': (email_settings['host'], 25),
        'fromaddr': email_settings['user'],
        'toaddrs': addrs,
        'subject': subject,
        'credentials': (email_settings['user'], email_settings['password'])
    }
    smtp_handler = logging.handlers.SMTPHandler(**handler_options)
    logger.addHandler(smtp_handler)
    logger.addHandler(logging.StreamHandler())
    return logger

admins = [item[1] for item in settings.ADMINS]
email_settings = {
    'user': settings.EMAIL_HOST_USER,
    'host': settings.EMAIL_HOST,
    'password': settings.EMAIL_HOST_PASSWORD,
}
logger = get_email_logger(admins, u"Replication error", email_settings)


grep = subprocess.Popen(
    ['ps', 'aux'],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE
)

out, error = grep.communicate()

try:
    if not ('wal receiver process' in out and 'streaming' in out):
        msg = 'no wal receiver proccess streaming data.'
        if 'waiting' in out:
            msg += ' recovery proccess is waiting for data.'
        raise Exception('Replication broken: ' + msg)
except:
    logger.error('Database replication error', exc_info=1)