Created
June 25, 2013 10:28
-
-
Save luke/5857483 to your computer and use it in GitHub Desktop.
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)
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment