Python logging handler for publishing to a slack channel.

slack_handler.py

import os
import time
import json
import socket
import logging

from slacker import Slacker, Error as SlackerError


class SlackChannelHandler(logging.Handler):
    """ Logging handler to send messages to a Slack Channel.

        Args:
            token (str)
            channel (str)

        Notes:
            All keyword arguments can be picked up as environment variables by prefixing the
            key with ``SLACK_`` and all-caps the parameter name afterwards. Boolean fields should
            be integers, ``0`` and ``1``.
    """

    COLORS = {
        'CRITICAL': '#DE5B49',
        'ERROR': '#E37B40',
        'WARN': '#F0CA4D',
        'WARNING': '#F0CA4D',
        'INFO': '#4180A8',
        'DEBUG': '#46B29D',
        'NOTSET': '#B2B2B2',
    }

    def __init__(self, token=None, channel=None, **extras):
        super(SlackChannelHandler, self).__init__()

        if extras.get('force_arguments', False):
            self.token = token
            self.channel = channel
        else:
            self.token = token or os.getenv('SLACK_TOKEN', token)
            self.channel = channel or os.getenv('SLACK_CHANNEL', channel)

        if not self.token:
            raise ValueError('must supply a slack-api token for sending messages')

        if not self.channel:
            raise ValueError('must supply a slack-channel to post these messages')

        self.channel = str(self.channel).strip().lstrip('#')

        self._slack = Slacker(self.token)

        channels = self._slack.channels.list(exclude_archived=True).body.get('channels', [])
        channels = [c['name'] for c in channels]

        if self.channel not in channels and not extras.get('private_channel', False):
            raise ValueError('cannot use channel %s, it does not exist' % self.channel)

        self._host = socket.gethostname()
        self._message_extras = {
            'channel': self.channel,
            'as_user': None,
            'link_names': None,
            'attachments': None,
            'username': 'Logger Jack',
            'icon_url': 'http://i.imgur.com/AbGYAf3.png',
            'icon_emoji': None,
            'mrkdwn': True
        }

        for k, v in self._message_extras.items():
            self._message_extras[k] = extras.get(k, v)

    def emit(self, record):
        out = {
            'fallback': self.format(record),
            'color': self.COLORS.get(record.levelname, self.COLORS['NOTSET']),
            'text': record.msg % record.args,
            'footer': '%s from %s, pid-%d' % (record.levelname, self._host, record.process),
            'ts': int(time.time())
        }

        self._message_extras['attachments'] = json.dumps([out])
        self._message_extras['text'] = '`{}`'.format(record.name)

        try:
            self._slack.chat.post('chat.postMessage', data=self._message_extras)
        except SlackerError:
            raise