AWS Firehose Handler

Python logging handler to load data into AWS Kinesis Delivery Stream

Demo Script

import logging.config

CONFIG = {
    'version': 1,
    'root': {
        'level': 'DEBUG',
        'handlers': ['console'],
    },
    'formatters': {
        'verbose': {
            'format': '%(asctime)s %(name)s %(levelname)s %(message)s %(pathname)s %(lineno)d %(funcName)s %(process)d %(processName)s %(thread)d %(threadName)s'
        },
        'json': {
            'class': 'pythonjsonlogger.jsonlogger.JsonFormatter',
            'format': '%(asctime)s %(name)s %(levelname)s %(message)s %(pathname)s %(lineno)d %(funcName)s %(process)d %(processName)s %(thread)d %(threadName)s'
        },
    },
    'handlers': {
        'console': {
            'level': 'DEBUG',
            'class': 'logging.StreamHandler',
            'formatter': 'verbose'
        },
        'firehose': {
            'level': 'INFO',
            'formatter': 'json',
            'class': 'firehose_handler.FirehoseHandler',
            'profile_name': 'Insert AWS Credential Profile Name',
            'region_name': 'Insert AWS Region Name',
            'delivery_stream_name': 'Insert Delivery Stream Name',
        },
    },
    'loggers': {
        'test-logger': {
            'handlers': ['firehose'],
            'level': 'INFO',
            'propagate': False,
        },
    },
}

logging.config.dictConfig(CONFIG)
logger = logging.getLogger('test-logger')


def test():
    try:
        raise NameError("fake NameError")
    except NameError as e:
        logger.error(e, exc_info=True)


if __name__ == '__main__':
    test()