Skip to content

Instantly share code, notes, and snippets.

@vivekhub
Forked from blakev/slack_handler.py
Created September 23, 2021 04:04
Show Gist options
  • Save vivekhub/2a166668372a4cc886ebc614933252b8 to your computer and use it in GitHub Desktop.
Save vivekhub/2a166668372a4cc886ebc614933252b8 to your computer and use it in GitHub Desktop.
Python logging handler for publishing to a slack channel.
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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment