Special log handler for use in Flask which sends emails using Amazon SES. Wraps stream handler using StringIO Configure with log level and trigger level (at which to send emails)

flask_ses_log_handler.py

#!/usr/bin/env python

import string, logging
from logging import StreamHandler
from flask import g
import boto
import StringIO

DEFAULT_FORMAT = "%(asctime)s %(levelname)-5s %(message)s"
DEFAULT_SUBJECT = "FlaskBufferingSESHandler log"
 
class StreamProxy:

    def __init__(self, proxy):
        self.proxy = proxy

    def write(self, data):
        self.proxy._stream.write(data) 

    def flush(self):
        self.proxy._stream.flush()

class FlaskSESLogHandler(StreamHandler):

    def __init__(self, app, 
        formatter=None, format=None, trigger_level=None, 
        source=None, to_addresses=None, subject=None, 
        ses=None, aws_access_key_id=None, aws_secret_access_key=None, **kwargs):

        # ARGS 

        format = format or app.config.get('SES_LOG_FORMAT') or DEFAULT_FORMAT
        source = source or app.config['SES_LOG_SOURCE']
        trigger_level = int(trigger_level or app.config.get('SES_TRIGGER_LEVEL') or logging.ERROR)
        to_addresses = to_addresses or app.config['SES_LOG_TO_ADDRESSES'].split(',')
        subject = subject or app.config.get('SES_LOG_SUBJECT') or DEFAULT_SUBJECT
        aws_access_key_id = aws_access_key_id or app.config.get('AWS_ACCESS_KEY_ID') 
        aws_secret_access_key = aws_secret_access_key or app.config.get('AWS_SECRET_ACCESS_KEY')  
               

        # LOGGING

        super(FlaskSESLogHandler, self).__init__(stream=StreamProxy(self))

        if formatter:
            self.setFormatter(formatter)
        else:
            self.setFormatter(logging.Formatter(format))

        self._trigger_level = trigger_level

        # APP 

        self._app = app

        _self = self

        @app.before_request
        def setup_log_handler():
            g._log_stream = StringIO.StringIO()
            g._log_triggered = False 

        @app.teardown_request
        def flush_log_handler(request):
            if g._log_triggered:
                _self.send_email()

        # SES

        self._ses = ses
        self._aws_access_key_id = aws_access_key_id 
        self._aws_secret_access_key = aws_secret_access_key

        kwargs['subject'] = subject
        kwargs['source'] = source 
        kwargs['to_addresses'] = to_addresses

        self._ses_send_email_kwargs = kwargs

        # do this manually to fix flask-ext standard
        # app.logger.addHandler(self) 
 
    @property
    def _stream(self):
        # called from StreamProxy
        return g._log_stream

    @property
    def ses(self):
        # lazy load ses 
        if not hasattr(self, '_ses') or self._ses is None:
            # NOTE: if the key and secret are none it will check env 
            self._ses = boto.connect_ses(self._aws_access_key_id, 
                self._aws_secret_access_key)
        return self._ses

    def emit(self, record):
        if record.levelno >= self._trigger_level:
            g._log_triggered = True
        super(FlaskSESLogHandler, self).emit(record)

    def flush(self): 
        pass # no nothing here

    def send_email(self):
        self.ses.send_email(body=self._stream.getvalue(), 
            **self._ses_send_email_kwargs)
 
# -----------------------------------------------------------------------------

import unittest

class MockSES:

    def send_email(self, **kwargs):
        self.sent_email = kwargs

def create_test_app(ses=None):
    from flask import Flask
    app = Flask(__name__)
    log_handler = FlaskSESLogHandler(app, 
        source="alerts@twilert.com", 
        to_addresses="luke@codegent.com",
        ses=ses)
    log_handler.setLevel(logging.DEBUG)
    app.debug = True
    app.logger.addHandler(log_handler)

    @app.route('/')
    def home():
        app.logger.debug("debug")
        app.logger.debug("debug")
        app.logger.debug("debug")
        app.logger.info("info")
        app.logger.warn("warn")
        app.logger.error("error!")
        try:
            ex = Exception("exception")
            raise ex 
        except Exception as ex:
            app.logger.exception(ex)
        return "testing logging"

    return app

class FlaskSESLogHandlerTestCase(unittest.TestCase):

    def setUp(self):
        self.mock_ses = MockSES()
        test_app = create_test_app(ses=self.mock_ses)
        test_app.config['TESTING'] = True
        self.app = test_app.test_client()

    def test_mock_ses(self):
        rv = self.app.get('/')
        assert 'testing logging' in rv.data
        assert 'error!' in self.mock_ses.sent_email['body']

    def tearDown(self):
        pass

if __name__ == '__main__':
    unittest.main()