diff --git a/README.md b/README.md index efe46d4..e27631b 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,7 @@ import os from slack_notifications import Slack + slack = Slack('') slack.send_notify('channel-name', username='Bot', text='@channel This is test message') ``` @@ -39,6 +40,7 @@ import os from slack_notifications import Slack, Attachment + slack = Slack('') message = slack.send_notify('channel-name', username='Bot', text='@channel This is test message') @@ -139,6 +141,137 @@ block = slack.SimpleTextBlock( slack.send_notify('channel-name', username='Bot', text='@channel This is test message', blocks=[block]) ``` +## Action Block + +```python +import slack_notifications as slack + + +slack.ACCESS_TOKEN = 'xxx' + + +block = slack.ActionsBlock( + elements=[ + slack.ButtonBlock( + 'Yes', + action_id='action1', + value='some_data1', + style='primary' + ), + slack.ButtonBlock( + 'No', + action_id='action2', + value='some_data2', + style='danger' + ), + ], +) + +slack.send_notify('channel-name', username='Bot', text='@channel This is test message', blocks=[block]) +``` + + +## Use mrkdwn module + +```python +import slack_notifications as slack + + +block = slack.SimpleTextBlock( + 'Text example', + fields=[ + slack.SimpleTextBlock.Field( + slack.mrkdwn.bold('Text field'), + ), + slack.SimpleTextBlock.Field( + slack.mrkdwn.italic('Text field'), + emoji=True, + ), + ], +) +``` + +## Mattermost interface + +### Simple usage + +```python +import os + +import slack_notifications.mattermost as mattermost + + +mattermost.ACCESS_TOKEN = 'xxx' +mattermost.BASE_URL_ENV_NAME = 'http://your-mattermost-url.com/api/v4' +mattermost.TEAM_ID_ENV_NAME = 'xxx' + +mattermost.send_notify('channel-name', username='Bot', text='@channel This is test message') +``` + +or + +```python +import os + +from slack_notifications.mattermost import Mattermost + + +mattermost = Mattermost('http://your-mattermost-url.com/api/v4', + token='', + team_id='xxx') +mattermost.send_notify('channel-name', username='Bot', text='@channel This is test message') +``` + + +### Use fields for Mattermost + +```python +import slack_notifications.mattermost as mattermost +import slack_notifications as slack + + +mattermost.ACCESS_TOKEN = 'xxx' +mattermost.BASE_URL = 'http://your-mattermost-url.com/api/v4' +mattermost.TEAM_ID = 'xxx' + + +block = slack.SimpleTextBlock( + 'Text example', + fields=[ + slack.SimpleTextBlock.Field( + 'Text field', + ), + slack.SimpleTextBlock.Field( + 'Text field', + emoji=True, + ), + ], +) + +mattermost.send_notify('channel-name', username='Bot', text='@channel This is test message', blocks=[block]) +``` + + +### Use mrkdwn module for Mattermost + +```python +import slack_notifications as slack +from slack_notifications.mattermost import mrkdwn + + +block = slack.SimpleTextBlock( + 'Text example', + fields=[ + slack.SimpleTextBlock.Field( + mrkdwn.bold('Text field'), + ), + slack.SimpleTextBlock.Field( + mrkdwn.italic('Text field'), + emoji=True, + ), + ], +) +``` See program API ## Init color @@ -300,8 +433,17 @@ slack.send_notify('channel-name', username='Bot', text='@channel This is test me * text: str * mrkdwn: bool = True - ### ContextBlock.ImageElement * image_url: str * alt_text: str = None + + +## ActionsBlock +* elements: List[ButtonBlock] + +### ButtonBlock +* text: str +* action_id: str +* value: str +* style: str = None diff --git a/setup.py b/setup.py index 60f11b5..64b85d1 100644 --- a/setup.py +++ b/setup.py @@ -1,6 +1,6 @@ from os import path -from setuptools import setup +from setuptools import setup, find_packages BASE_PATH = path.abspath(path.dirname(__file__)) @@ -9,6 +9,7 @@ setup( name='slack-notifications', + packages=find_packages(include=('slack_notifications', 'slack_notifications.*')), version_format='{tag}', setup_requires=['setuptools-git-version'], description='Send notifications to slack channel with supporting attachments and fields', diff --git a/slack_notifications.py b/slack_notifications.py deleted file mode 100644 index b621ab4..0000000 --- a/slack_notifications.py +++ /dev/null @@ -1,641 +0,0 @@ -import os -import string -import random -import logging -from typing import List, Union - -import requests - -logger = logging.getLogger(__name__) - -ACCESS_TOKEN = None -ACCESS_TOKEN_ENV_NAME = 'SLACK_ACCESS_TOKEN' - -COLOR_MAP = { - 'green': '#008000', - 'gray': '#808080', - 'red': '#FF0000', - 'blue': '#0000FF', - 'black': '#000000', - 'yellow': '#FFFF00', - 'maroon': '#800000', - 'purple': '#800080', - 'olive': '#808000', - 'silver': '#C0C0C0', - 'gold': '#FFD700', - 'pink': '#FFC0CB', - 'coral': '#FF7F50', - 'brown': '#A52A2A', - 'indigo': '#4B0082', - 'aqua': '#00FFFF', - 'cyan': '#00FFFF', - 'lime': '#00FF00', - 'teal': '#008080', - 'navy': '#000080', - 'sienna': '#A0522D', -} - - -def _random_string(length): - letters = string.ascii_lowercase - return ''.join(random.choice(letters) for i in range(length)) - - -class SlackError(requests.exceptions.RequestException): - pass - - -class Resource: - - def __init__(self, handle: str, method: str): - self.handle = handle - self.method = method - - -class DictConvertibleObject: - - def __init__(self, *args, **kwargs): - pass - - def to_dict(self): - raise NotImplementedError( - 'Object "{}" does not implemented "to_dict" method'.format(self.__class__.__name__), - ) - - -class AttachmentField(DictConvertibleObject): - - def __init__(self, *, title: str = None, value: str = None, short: bool = False): - super(AttachmentField, self).__init__() - - self.title = title - self.value = value - self.short = short - - def to_dict(self): - assert self.title is not None or self.value is not None, \ - 'Title or value is required for attachment field' - - data = {'short': self.short} - - if self.title: - data['title'] = self.title - - if self.value: - data['value'] = self.value - - return data - - -class Attachment(DictConvertibleObject): - Field = AttachmentField - - def __init__(self, *, - image_url: str = None, - thumb_url: str = None, - author_name: str = None, - author_link: str = None, - author_icon: str = None, - title: str = None, - title_link: str = None, - text: str = None, - pretext: str = None, - footer: str = None, - footer_icon: str = None, - timestamp: str = None, - fields: List[AttachmentField] = None, - mrkdwn: bool = True, - color: str = None): - super(Attachment, self).__init__() - - self.image_url = image_url - self.thumb_url = thumb_url - - self.author_name = author_name - self.author_link = author_link - self.author_icon = author_icon - - self.title = title - self.title_link = title_link - - self.text = text - - self.pretext = pretext - - self.footer = footer - self.footer_icon = footer_icon - - self.timestamp = timestamp - - self.fields = fields - - self.mrkdwn = mrkdwn - self.color = color - - def to_dict(self): - data = { - 'mrkdwn_in': [], - } - - if self.color: - data['color'] = COLOR_MAP.get(self.color, self.color) - - if self.image_url: - data['image_url'] = self.image_url - - if self.thumb_url: - data['thumb_url'] = self.thumb_url - - if self.author_name: - data['author_name'] = self.author_name - - if self.author_link: - data['author_link'] = self.author_link - - if self.author_icon: - data['author_icon'] = self.author_icon - - if self.title: - data['title'] = self.title - if self.mrkdwn: - data['mrkdwn_in'].append('title') - - if self.title_link: - data['title_link'] = self.title_link - - if self.pretext: - data['pretext'] = self.pretext - if self.mrkdwn: - data['mrkdwn_in'].append('pretext') - - if self.text: - data['text'] = self.text - if self.mrkdwn: - data['mrkdwn_in'].append('text') - - if self.footer: - data['footer'] = self.footer - if self.mrkdwn: - data['mrkdwn_in'].append('footer') - - if self.footer_icon: - data['footer_icon'] = self.footer_icon - - if self.timestamp: - data['ts'] = self.timestamp - - if self.fields: - data['fields'] = [f.to_dict() for f in self.fields] - if self.mrkdwn: - data['mrkdwn_in'].append('fields') - - return data - - -class BaseBlock(DictConvertibleObject): - __type__ = None - - def __init__(self, *, mrkdwn: bool = True, block_id: str = None): - super(BaseBlock, self).__init__() - - self.mrkdwn = mrkdwn - self.block_id = block_id - self.content_type = 'mrkdwn' if self.mrkdwn else 'plain_text' - - def to_dict(self): - data = { - 'type': self.__type__, - } - - if self.block_id: - data['block_id'] = self.block_id - - return data - - -class BaseBlockField(DictConvertibleObject): - __type__ = None - - def __init__(self, *, mrkdwn=True): - super(BaseBlockField, self).__init__() - - self.mrkdwn = mrkdwn - self.content_type = 'mrkdwn' if self.mrkdwn else 'plain_text' - - def to_dict(self): - if self.__type__: - return { - 'type': self.__type__, - } - - return {} - - -class HeaderBlock(BaseBlock): - __type__ = 'header' - - def __init__(self, text: str, **kwargs): - kwargs['mrkdwn'] = False - super().__init__(**kwargs) - - self.text = text - - def to_dict(self): - data = super().to_dict() - - data['text'] = { - 'type': self.content_type, - 'text': self.text, - } - - return data - - -class SimpleTextBlockField(BaseBlockField): - - def __init__(self, text: str, *, emoji: bool = None, **kwargs): - super(SimpleTextBlockField, self).__init__(**kwargs) - - self.text = text - self.emoji = emoji - - def to_dict(self): - data = super(SimpleTextBlockField, self).to_dict() - - data['text'] = self.text - data['type'] = self.content_type - - if self.emoji is not None: - data['emoji'] = self.emoji - - return data - - -class SimpleTextBlock(BaseBlock): - __type__ = 'section' - - Field = SimpleTextBlockField - - def __init__(self, text: str, *, fields: List[SimpleTextBlockField] = None, **kwargs): - super(SimpleTextBlock, self).__init__(**kwargs) - - self.text = text - self.fields = fields - - def to_dict(self): - data = super(SimpleTextBlock, self).to_dict() - - data['text'] = { - 'type': self.content_type, - 'text': self.text, - } - - if self.fields: - data['fields'] = [f.to_dict() for f in self.fields] - - return data - - -class DividerBlock(BaseBlock): - __type__ = 'divider' - - -class ImageBlock(BaseBlock): - __type__ = 'image' - - def __init__(self, image_url, *, title: str = None, alt_text: str = None, **kwargs): - super(ImageBlock, self).__init__(**kwargs) - - self.image_url = image_url - - self.title = title - self.alt_text = alt_text or image_url - - def to_dict(self): - data = super(ImageBlock, self).to_dict() - - data['image_url'] = self.image_url - - if self.title: - data['title'] = { - 'type': self.content_type, - 'text': self.title, - } - - if self.alt_text: - data['alt_text'] = self.alt_text - - return data - - -class ContextBlockTextElement(BaseBlockField): - - def __init__(self, text, **kwargs): - super(ContextBlockTextElement, self).__init__(**kwargs) - - self.text = text - - def to_dict(self): - data = super(ContextBlockTextElement, self).to_dict() - - data['text'] = self.text - data['type'] = self.content_type - - return data - - -class ContextBlockImageElement(BaseBlockField): - __type__ = 'image' - - def __init__(self, image_url, alt_text: str = None): - super(ContextBlockImageElement, self).__init__() - - self.image_url = image_url - self.alt_text = alt_text - - def to_dict(self): - data = super(ContextBlockImageElement, self).to_dict() - - data['image_url'] = self.image_url - - if self.alt_text: - data['alt_text'] = self.alt_text - - return data - - -class ContextBlock(BaseBlock): - __type__ = 'context' - - TextElement = ContextBlockTextElement - ImageElement = ContextBlockImageElement - - def __init__(self, elements: List[Union[ContextBlockTextElement, ContextBlockImageElement]], **kwargs): - super(ContextBlock, self).__init__(**kwargs) - - self.elements = elements - - def to_dict(self): - data = super(ContextBlock, self).to_dict() - - data['elements'] = [e.to_dict() for e in self.elements] - - return data - - -def init_color(name, code): - COLOR_MAP[name] = code - - -class Message: - - def __init__(self, client, response, - text: str = None, - raise_exc=False, - attachments: List[Attachment] = None, - blocks: List[BaseBlock] = None): - self._client = client - self._response = response - self._raise_exc = raise_exc - - self.text = text - self.attachments = attachments or [] - self.blocks = blocks or [] - - self.__lock_thread = False - - @property - def response(self): - return self._response - - def _lock_thread(self): - self.__lock_thread = True - - def add_reaction(self, name, raise_exc=False): - json = self._response.json() - data = { - 'name': name, - 'channel': json['channel'], - 'timestamp': json['message']['ts'], - } - return self._client.call_resource( - Resource('reactions.add', 'POST'), - json=data, raise_exc=raise_exc, - ) - - def remove_reaction(self, name, raise_exc=False): - json = self._response.json() - data = { - 'name': name, - 'channel': json['channel'], - 'timestamp': json['message']['ts'], - } - return self._client.call_resource( - Resource('reactions.remove', 'POST'), - json=data, raise_exc=raise_exc, - ) - - def send_to_thread(self, **kwargs): - if self.__lock_thread: - raise SlackError('Cannot open thread for thread message') - - json = self._response.json() - thread_ts = json['message']['ts'] - kwargs.update(thread_ts=thread_ts) - - message = self._client.send_notify(json['channel'], **kwargs) - - lock_thread = getattr(message, '_lock_thread') - lock_thread() - - return message - - def update(self): - json = self._response.json() - data = { - 'channel': json['channel'], - 'ts': json['message']['ts'], - } - - if self.text: - data['text'] = self.text - - if self.blocks: - data['blocks'] = [b.to_dict() for b in self.blocks] - - if self.attachments: - data['attachments'] = [a.to_dict() for a in self.attachments] - - return self._client.call_resource( - Resource('chat.update', 'POST'), - json=data, raise_exc=self._raise_exc, - ) - - def delete(self): - json = self._response.json() - data = { - 'channel': json['channel'], - 'ts': json['message']['ts'], - } - return self._client.call_resource( - Resource('chat.update', 'POST'), - json=data, raise_exc=self._raise_exc, - ) - - def upload_file(self, file, **kwargs): - json = self._response.json() - kwargs.update(thread_ts=json['message']['ts']) - return self._client.upload_file(json['channel'], file, **kwargs) - - -class Slack(requests.Session): - API_URL = 'https://slack.com/api' - - DEFAULT_RECORDS_LIMIT = 100 - DEFAULT_REQUEST_TIMEOUT = 180 - - def __init__(self, token): - super(Slack, self).__init__() - - self.headers['Authorization'] = 'Bearer {}'.format(token) - self.headers['Content-Type'] = 'application/json; charset=utf-8' - - @classmethod - def from_env(cls): - token = ACCESS_TOKEN or os.getenv(ACCESS_TOKEN_ENV_NAME) - assert token is not None, 'Please export "{}" environment variable'.format(ACCESS_TOKEN_ENV_NAME) - return cls(token) - - def call_resource(self, resource: Resource, *, raise_exc: bool = False, **kwargs): - kwargs.setdefault('timeout', self.DEFAULT_REQUEST_TIMEOUT) - - url = '{}/{}'.format(self.API_URL, resource.handle) - response = self.request(resource.method, url, **kwargs) - - logger.debug(response.content) - - if raise_exc: - response.raise_for_status() - - json = response.json() - - if not json['ok']: - logger.error(response.content) - raise SlackError(response.content) - - return response - - def resource_iterator(self, - resource: Resource, from_key: str, *, - cursor: str = None, - raise_exc: bool = False, - limit: int = None): - params = {'limit': limit} - - if cursor: - params['cursor'] = cursor - - response = self.call_resource(resource, params=params, raise_exc=raise_exc) - data = response.json() - - for item in data[from_key]: - yield item - - cursor = data.get('response_metadata', {}).get('next_cursor') - - if cursor: - yield from self.resource_iterator( - resource, from_key, - limit=limit or self.DEFAULT_RECORDS_LIMIT, cursor=cursor, raise_exc=raise_exc, - ) - - def upload_file(self, - channel, file, *, - title: str = None, - content: str = None, - filename: str = None, - thread_ts: str = None, - filetype: str = 'text', - raise_exc: bool = False): - data = { - 'channels': channel, - 'filetype': filetype, - } - if isinstance(file, str) and content: - filename = file - data.update(content=content, filename=filename) - elif isinstance(file, str) and not content: - data.update(filename=os.path.basename(file)) - with open(file, 'r') as f: - data.update(content=f.read()) - else: - data.update(content=file.read(), filename=filename or _random_string(7)) - - if title: - data.update(title=title) - if thread_ts: - data.update(thread_ts=thread_ts) - - return self.call_resource( - Resource('files.upload', 'POST'), - data=data, - raise_exc=raise_exc, - headers={ - 'Content-Type': 'application/x-www-form-urlencoded', - }, - ) - - def send_notify(self, - channel, *, - text: str = None, - username: str = None, - icon_url: str = None, - icon_emoji: str = None, - link_names: bool = True, - raise_exc: bool = False, - attachments: List[Attachment] = None, - blocks: List[BaseBlock] = None, - thread_ts: str = None): - data = { - 'channel': channel, - 'link_names': link_names, - } - - if username: - data['username'] = username - - if text: - data['mrkdwn'] = True - data['text'] = text - - if icon_url: - data['icon_url'] = icon_url - - if icon_emoji: - data['icon_emoji'] = icon_emoji - - if blocks: - data['blocks'] = [b.to_dict() for b in blocks] - - if attachments: - data['attachments'] = [a.to_dict() for a in attachments] - - if thread_ts: - data['thread_ts'] = thread_ts - - response = self.call_resource( - Resource('chat.postMessage', 'POST'), raise_exc=raise_exc, json=data, - ) - return Message(self, response, text=text, raise_exc=raise_exc, blocks=blocks, attachments=attachments) - - -def call_resource(*args, **kwargs): - return Slack.from_env().call_resource(*args, **kwargs) - - -def resource_iterator(*args, **kwargs): - return Slack.from_env().resource_iterator(*args, **kwargs) - - -def send_notify(*args, **kwargs): - return Slack.from_env().send_notify(*args, **kwargs) diff --git a/slack_notifications/__init__.py b/slack_notifications/__init__.py new file mode 100644 index 0000000..4090b78 --- /dev/null +++ b/slack_notifications/__init__.py @@ -0,0 +1,60 @@ +from slack_notifications.constants import COLOR_MAP +from slack_notifications.utils import init_color +from slack_notifications.common import Resource +from slack_notifications.slack import ( + Slack, + Message, + call_resource, + resource_iterator, + send_notify, + ACCESS_TOKEN +) +from slack_notifications.fields import mrkdwn +from slack_notifications.fields.blocks import ( + BaseBlock, + BaseBlockField, + HeaderBlock, + SimpleTextBlockField, + SimpleTextBlock, + DividerBlock, + ImageBlock, + ContextBlock, + ContextBlockTextElement, + ContextBlockImageElement, + ActionsBlock, + ButtonBlock +) +from slack_notifications.fields.attachments import ( + Attachment, + AttachmentField +) +from slack_notifications import mattermost + + +__all__ = [ + 'ACCESS_TOKEN', + 'COLOR_MAP', + 'init_color', + 'Resource', + 'Slack', + 'Message', + 'call_resource', + 'resource_iterator', + 'send_notify', + 'BaseBlock', + 'BaseBlockField', + 'HeaderBlock', + 'SimpleTextBlockField', + 'SimpleTextBlock', + 'DividerBlock', + 'ImageBlock', + 'ContextBlock', + 'ContextBlockTextElement', + 'ContextBlockImageElement', + 'ActionsBlock', + 'ButtonBlock', + 'Attachment', + 'AttachmentField', + 'mrkdwn', + 'mattermost', +] diff --git a/slack_notifications/common.py b/slack_notifications/common.py new file mode 100644 index 0000000..13b3c39 --- /dev/null +++ b/slack_notifications/common.py @@ -0,0 +1,55 @@ +import requests +import logging + +logger = logging.getLogger(__name__) + + +class DictConvertibleObject: + + def __init__(self, *args, **kwargs): + pass + + def to_dict(self): + raise NotImplementedError( + 'Object "{}" does not implemented "to_dict" method'.format(self.__class__.__name__), + ) + + +class NotificationError(requests.exceptions.RequestException): + pass + + +class Resource: + def __init__(self, handle: str, method: str): + self.handle = handle + self.method = method + + +class NotificationClient(requests.Session): + DEFAULT_RECORDS_LIMIT = 100 + DEFAULT_REQUEST_TIMEOUT = 180 + + def __init__(self, base_url, *, token): + self.base_url = base_url + self._token = token + super().__init__() + + self.headers['Authorization'] = 'Bearer {}'.format(token) + self.headers['Content-Type'] = 'application/json; charset=utf-8' + + def call_resource(self, resource: Resource, *, raise_exc: bool = False, **kwargs): + kwargs.setdefault('timeout', self.DEFAULT_REQUEST_TIMEOUT) + + url = '{}/{}'.format(self.base_url, resource.handle) + response = self.request(resource.method, url, **kwargs) + + logger.debug(response.content) + + if raise_exc: + response.raise_for_status() + + if not response.ok: + logger.error(response.content) + raise NotificationError(response.content) + + return response diff --git a/slack_notifications/constants.py b/slack_notifications/constants.py new file mode 100644 index 0000000..88f57bb --- /dev/null +++ b/slack_notifications/constants.py @@ -0,0 +1,23 @@ +COLOR_MAP = { + 'green': '#008000', + 'gray': '#808080', + 'red': '#FF0000', + 'blue': '#0000FF', + 'black': '#000000', + 'yellow': '#FFFF00', + 'maroon': '#800000', + 'purple': '#800080', + 'olive': '#808000', + 'silver': '#C0C0C0', + 'gold': '#FFD700', + 'pink': '#FFC0CB', + 'coral': '#FF7F50', + 'brown': '#A52A2A', + 'indigo': '#4B0082', + 'aqua': '#00FFFF', + 'cyan': '#00FFFF', + 'lime': '#00FF00', + 'teal': '#008080', + 'navy': '#000080', + 'sienna': '#A0522D', +} diff --git a/slack_notifications/fields/__init__.py b/slack_notifications/fields/__init__.py new file mode 100644 index 0000000..1b95ae3 --- /dev/null +++ b/slack_notifications/fields/__init__.py @@ -0,0 +1,38 @@ +from slack_notifications.fields.blocks import ( + BaseBlock, + BaseBlockField, + HeaderBlock, + SimpleTextBlockField, + SimpleTextBlock, + DividerBlock, + ImageBlock, + ContextBlock, + ContextBlockTextElement, + ContextBlockImageElement, + ActionsBlock, + ButtonBlock +) +from slack_notifications.fields.attachments import ( + Attachment, + AttachmentField +) +from slack_notifications.fields import mrkdwn + + +__all__ = [ + 'BaseBlock', + 'BaseBlockField', + 'HeaderBlock', + 'SimpleTextBlockField', + 'SimpleTextBlock', + 'DividerBlock', + 'ImageBlock', + 'ContextBlock', + 'ContextBlockTextElement', + 'ContextBlockImageElement', + 'ActionsBlock', + 'ButtonBlock', + 'Attachment', + 'AttachmentField', + 'mrkdwn', +] diff --git a/slack_notifications/fields/attachments.py b/slack_notifications/fields/attachments.py new file mode 100644 index 0000000..15b5b8d --- /dev/null +++ b/slack_notifications/fields/attachments.py @@ -0,0 +1,133 @@ +from typing import List + +from slack_notifications.common import DictConvertibleObject +from slack_notifications.constants import COLOR_MAP + + +class AttachmentField(DictConvertibleObject): + + def __init__(self, *, title: str = None, value: str = None, short: bool = False): + super(AttachmentField, self).__init__() + + self.title = title + self.value = value + self.short = short + + def to_dict(self): + assert self.title is not None or self.value is not None, \ + 'Title or value is required for attachment field' + + data = {'short': self.short} + + if self.title: + data['title'] = self.title + + if self.value: + data['value'] = self.value + + return data + + +class Attachment(DictConvertibleObject): + Field = AttachmentField + + def __init__(self, *, + image_url: str = None, + thumb_url: str = None, + author_name: str = None, + author_link: str = None, + author_icon: str = None, + title: str = None, + title_link: str = None, + text: str = None, + pretext: str = None, + footer: str = None, + footer_icon: str = None, + timestamp: str = None, + fields: List[AttachmentField] = None, + mrkdwn: bool = True, + color: str = None): + super(Attachment, self).__init__() + + self.image_url = image_url + self.thumb_url = thumb_url + + self.author_name = author_name + self.author_link = author_link + self.author_icon = author_icon + + self.title = title + self.title_link = title_link + + self.text = text + + self.pretext = pretext + + self.footer = footer + self.footer_icon = footer_icon + + self.timestamp = timestamp + + self.fields = fields + + self.mrkdwn = mrkdwn + self.color = color + + def to_dict(self): + data = { + 'mrkdwn_in': [], + } + + if self.color: + data['color'] = COLOR_MAP.get(self.color, self.color) + + if self.image_url: + data['image_url'] = self.image_url + + if self.thumb_url: + data['thumb_url'] = self.thumb_url + + if self.author_name: + data['author_name'] = self.author_name + + if self.author_link: + data['author_link'] = self.author_link + + if self.author_icon: + data['author_icon'] = self.author_icon + + if self.title: + data['title'] = self.title + if self.mrkdwn: + data['mrkdwn_in'].append('title') + + if self.title_link: + data['title_link'] = self.title_link + + if self.pretext: + data['pretext'] = self.pretext + if self.mrkdwn: + data['mrkdwn_in'].append('pretext') + + if self.text: + data['text'] = self.text + if self.mrkdwn: + data['mrkdwn_in'].append('text') + + if self.footer: + data['footer'] = self.footer + if self.mrkdwn: + data['mrkdwn_in'].append('footer') + + if self.footer_icon: + data['footer_icon'] = self.footer_icon + + if self.timestamp: + data['ts'] = self.timestamp + + if self.fields: + data['fields'] = [f.to_dict() for f in self.fields] + if self.mrkdwn: + data['mrkdwn_in'].append('fields') + + return data diff --git a/slack_notifications/fields/blocks.py b/slack_notifications/fields/blocks.py new file mode 100644 index 0000000..c903be9 --- /dev/null +++ b/slack_notifications/fields/blocks.py @@ -0,0 +1,236 @@ +from typing import List, Union + +from slack_notifications.common import DictConvertibleObject + + +class BaseBlock(DictConvertibleObject): + __type__ = None + + def __init__(self, *, mrkdwn: bool = True, block_id: str = None): + super(BaseBlock, self).__init__() + + self.mrkdwn = mrkdwn + self.block_id = block_id + self.content_type = 'mrkdwn' if self.mrkdwn else 'plain_text' + + def to_dict(self): + data = { + 'type': self.__type__, + } + + if self.block_id: + data['block_id'] = self.block_id + + return data + + +class BaseBlockField(DictConvertibleObject): + __type__ = None + + def __init__(self, *, mrkdwn=True): + super(BaseBlockField, self).__init__() + + self.mrkdwn = mrkdwn + self.content_type = 'mrkdwn' if self.mrkdwn else 'plain_text' + + def to_dict(self): + if self.__type__: + return { + 'type': self.__type__, + } + + return {} + + +class HeaderBlock(BaseBlock): + __type__ = 'header' + + def __init__(self, text: str, **kwargs): + kwargs['mrkdwn'] = False + super().__init__(**kwargs) + + self.text = text + + def to_dict(self): + data = super().to_dict() + + data['text'] = { + 'type': self.content_type, + 'text': self.text, + } + + return data + + +class SimpleTextBlockField(BaseBlockField): + + def __init__(self, text: str, *, emoji: bool = None, **kwargs): + super(SimpleTextBlockField, self).__init__(**kwargs) + + self.text = text + self.emoji = emoji + + def to_dict(self): + data = super(SimpleTextBlockField, self).to_dict() + + data['text'] = self.text + data['type'] = self.content_type + + if self.emoji is not None: + data['emoji'] = self.emoji + + return data + + +class SimpleTextBlock(BaseBlock): + __type__ = 'section' + + Field = SimpleTextBlockField + + def __init__(self, text: str, *, fields: List[SimpleTextBlockField] = None, **kwargs): + super(SimpleTextBlock, self).__init__(**kwargs) + + self.text = text + self.fields = fields + + def to_dict(self): + data = super(SimpleTextBlock, self).to_dict() + + data['text'] = { + 'type': self.content_type, + 'text': self.text, + } + + if self.fields: + data['fields'] = [f.to_dict() for f in self.fields] + + return data + + +class DividerBlock(BaseBlock): + __type__ = 'divider' + + +class ImageBlock(BaseBlock): + __type__ = 'image' + + def __init__(self, image_url, *, title: str = None, alt_text: str = None, **kwargs): + super(ImageBlock, self).__init__(**kwargs) + + self.image_url = image_url + + self.title = title + self.alt_text = alt_text or image_url + + def to_dict(self): + data = super(ImageBlock, self).to_dict() + + data['image_url'] = self.image_url + + if self.title: + data['title'] = { + 'type': self.content_type, + 'text': self.title, + } + + if self.alt_text: + data['alt_text'] = self.alt_text + + return data + + +class ContextBlockTextElement(BaseBlockField): + + def __init__(self, text, **kwargs): + super(ContextBlockTextElement, self).__init__(**kwargs) + + self.text = text + + def to_dict(self): + data = super(ContextBlockTextElement, self).to_dict() + + data['text'] = self.text + data['type'] = self.content_type + + return data + + +class ContextBlockImageElement(BaseBlockField): + __type__ = 'image' + + def __init__(self, image_url, alt_text: str = None): + super(ContextBlockImageElement, self).__init__() + + self.image_url = image_url + self.alt_text = alt_text + + def to_dict(self): + data = super(ContextBlockImageElement, self).to_dict() + + data['image_url'] = self.image_url + + if self.alt_text: + data['alt_text'] = self.alt_text + + return data + + +class ContextBlock(BaseBlock): + __type__ = 'context' + + TextElement = ContextBlockTextElement + ImageElement = ContextBlockImageElement + + def __init__(self, elements: List[Union[ContextBlockTextElement, ContextBlockImageElement]], **kwargs): + super(ContextBlock, self).__init__(**kwargs) + + self.elements = elements + + def to_dict(self): + data = super(ContextBlock, self).to_dict() + + data['elements'] = [e.to_dict() for e in self.elements] + + return data + + +class ButtonBlock(BaseBlock): + __type__ = 'button' + + def __init__(self, text: str, *, action_id: str, value: str, style: str = None, **kwargs): + super(ButtonBlock, self).__init__(**kwargs) + + self.text = text + self.action_id = action_id + self.value = value + self.style = style + + def to_dict(self): + data = super(ButtonBlock, self).to_dict() + + data['action_id'] = self.action_id + data['value'] = self.value + data['text'] = { + 'type': 'plain_text', + 'text': self.text, + } + if self.style is not None: + data['style'] = self.style + + return data + + +class ActionsBlock(BaseBlock): + __type__ = 'actions' + + def __init__(self, *, elements: List[ButtonBlock], **kwargs): + super(ActionsBlock, self).__init__(**kwargs) + + self.elements = elements + + def to_dict(self): + data = super(ActionsBlock, self).to_dict() + + data['elements'] = [e.to_dict() for e in self.elements] + + return data diff --git a/slack_notifications/fields/mrkdwn.py b/slack_notifications/fields/mrkdwn.py new file mode 100644 index 0000000..01b0ac4 --- /dev/null +++ b/slack_notifications/fields/mrkdwn.py @@ -0,0 +1,41 @@ +from typing import Iterable + + +def bold(text): + return f'*{str(text)}*' + + +def italic(text): + return f'_{str(text)}_' + + +def strike(text): + return f'~{str(text)}~' + + +def code_block(text): + return f'`{str(text)}`' + + +def multi_line_code_block(text): + return f'```{str(text)}```' + + +def block_quotes(text): + return f'\n>{str(text)}\n' + + +def list_items(itr: Iterable): + return '\n'.join([f'• {str(item)}' for item in itr]) + '\n' + + +def link(url, text): + return f'<{url}|{str(text)}>' + + +def user_mention(user_id): + return f'<@{user_id}>' + + +def channel_mention(channel): + return f'<#{channel}>' diff --git a/slack_notifications/interfaces/__init__.py b/slack_notifications/interfaces/__init__.py new file mode 100644 index 0000000..5005c91 --- /dev/null +++ b/slack_notifications/interfaces/__init__.py @@ -0,0 +1,7 @@ +from slack_notifications.interfaces.client import InterfaceMessage, InterfaceClient + + +__all__ = [ + 'InterfaceMessage', + 'InterfaceClient', +] diff --git a/slack_notifications/interfaces/client.py b/slack_notifications/interfaces/client.py new file mode 100644 index 0000000..a31b684 --- /dev/null +++ b/slack_notifications/interfaces/client.py @@ -0,0 +1,50 @@ +from typing import Protocol, List + +from slack_notifications.fields.blocks import BaseBlock +from slack_notifications.fields.attachments import Attachment + + +class InterfaceMessage(Protocol): + + def send_to_thread(self, **kwargs): + pass + + def update(self): + pass + + def delete(self): + pass + + def upload_file(self, file, **kwargs): + pass + + def add_reaction(self, name, raise_exc=False): + pass + + def remove_reaction(self, name, raise_exc=False): + pass + + +class InterfaceClient(Protocol): + def send_notify(self, + channel, *, + text: str = None, + username: str = None, + icon_url: str = None, + icon_emoji: str = None, + link_names: bool = True, + raise_exc: bool = False, + attachments: List[Attachment] = None, + blocks: List[BaseBlock] = None, + thread_ts: str = None) -> InterfaceMessage: + pass + + def upload_file(self, + channel, file, *, + title: str = None, + content: str = None, + filename: str = None, + thread_ts: str = None, + filetype: str = 'text', + raise_exc: bool = False): + pass diff --git a/slack_notifications/mattermost/__init__.py b/slack_notifications/mattermost/__init__.py new file mode 100644 index 0000000..c0909ae --- /dev/null +++ b/slack_notifications/mattermost/__init__.py @@ -0,0 +1,24 @@ +from slack_notifications.mattermost.client import ( + Mattermost, + MattermostMessage as Message, + call_resource, + send_notify, + ACCESS_TOKEN, + BASE_URL, + TEAM_ID +) +from slack_notifications.mattermost.fields import ( + mrkdwn +) + + +__all__ = [ + 'ACCESS_TOKEN', + 'BASE_URL', + 'TEAM_ID', + 'Mattermost', + 'Message', + 'call_resource', + 'send_notify', + 'mrkdwn', +] diff --git a/slack_notifications/mattermost/client.py b/slack_notifications/mattermost/client.py new file mode 100644 index 0000000..7e57ba0 --- /dev/null +++ b/slack_notifications/mattermost/client.py @@ -0,0 +1,254 @@ +import os +from typing import List +import logging +import requests + +from slack_notifications.utils import _random_string +from slack_notifications.common import NotificationClient, Resource +from slack_notifications.fields.blocks import BaseBlock +from slack_notifications.fields.attachments import Attachment +from slack_notifications.mattermost.converter import MattermostConverter + +ACCESS_TOKEN = None +ACCESS_TOKEN_ENV_NAME = 'MATTERMOST_ACCESS_TOKEN' +BASE_URL = None +BASE_URL_ENV_NAME = 'MATTERMOST_URL' +TEAM_ID = None +TEAM_ID_ENV_NAME = 'MATTERMOST_TEAM_ID_TOKEN' + +logger = logging.getLogger(__name__) + + +class MattermostError(requests.exceptions.RequestException): + pass + + +class MattermostMessage: + def __init__(self, client, response, + text: str = None, + raise_exc=False, + attachments: List = None, + blocks: List = None): + self._client = client + self._response = response + self._raise_exc = raise_exc + + self.text = text + self.attachments = attachments or [] + self.blocks = blocks or [] + + @property + def response(self): + return self._response + + def send_to_thread(self, **kwargs): + json = self._response.json() + message_id = json['id'] + kwargs.update(thread_ts=message_id) + + message = self._client.send_notify(json['channel_id'], **kwargs) + + return message + + def update(self): + json = self._response.json() + message_id = json['id'] + data = { + 'id': json['id'], + 'message': self.text or '', + 'props': {}, + 'metadata': {} + } + converter = MattermostConverter() + converter.convert(blocks=self.blocks) + + if self.blocks: + data['message'] += ''.join(map(str, self.blocks)) + + data['props']['attachments'] = [] + if converter.attachments_result: + data['props']['attachments'].extend(converter.attachments_result) + if self.attachments: + data['props']['attachments'].extend([a.to_dict() for a in self.attachments]) + + return self._client.call_resource( + Resource(f'posts/{message_id}/patch', 'PUT'), + json=data, raise_exc=self._raise_exc, + ) + + def delete(self): + message_id = self._response.json()['id'] + response = self._client.call_resource( + Resource(f'posts/{message_id}', 'DELETE'), + raise_exc=self._raise_exc + ) + return response + + def upload_file(self, file, **kwargs): + json = self._response.json() + message_id = json['id'] + kwargs.update(thread_ts=message_id) + return self._client.upload_file(json['channel_id'], file, **kwargs) + + def add_reaction(self, name, raise_exc=False): + json = self._response.json() + data = { + 'emoji_name': name, + 'post_id': json['id'], + 'user_id': json['user_id'], + 'create_at': json['create_at'] + } + return self._client.call_resource( + Resource('reactions', 'POST'), + json=data, raise_exc=raise_exc, + ) + + def remove_reaction(self, name, raise_exc=False): + json = self._response.json() + user_id = json['user_id'] + post_id = json['post_id'] + + return self._client.call_resource( + Resource(f'users/{user_id}/posts/{post_id}/reactions/{name}', 'DELETE'), + raise_exc=raise_exc, + ) + + +class Mattermost(NotificationClient): + + def __init__(self, base_url, *, token, team_id=None): + super(Mattermost, self).__init__(base_url, token=token) + self._team_id = team_id + + def _is_channel_id(self, channel): + if len(channel) == 26 and channel.isalnum(): + return True + + def channel_id_by_name(self, channel_name): + response = self.call_resource(Resource(f'teams/{self._team_id}/channels/name/{channel_name}', 'GET')) + + if response.status_code != 200: + raise ValueError('Channel not found') + + return response.json()['id'] + + def set_team_id_by_name(self, team_name): + response = self.call_resource(Resource('teams', 'GET')) + json = response.json() + for item in json: + if item['name'] == team_name: + self._team_id = item['id'] + break + + @classmethod + def from_env(cls): + token = ACCESS_TOKEN or os.getenv(ACCESS_TOKEN_ENV_NAME) + base_url = BASE_URL or os.getenv(BASE_URL_ENV_NAME) + team_id = TEAM_ID or os.getenv(TEAM_ID_ENV_NAME) + + return cls(base_url, token=token, team_id=team_id) + + def send_notify(self, + channel, *, + text: str = None, + username: str = None, + icon_url: str = None, + icon_emoji: str = None, + link_names: bool = True, + raise_exc: bool = False, + attachments: List[Attachment] = None, + blocks: List[BaseBlock] = None, + thread_ts: str = None): + if not self._is_channel_id(channel): + channel = self.channel_id_by_name(channel) + + data = { + 'channel_id': channel, + 'message': text or '', + 'props': {}, + 'metadata': {} + } + converter = MattermostConverter() + converter.convert(blocks=blocks) + + if blocks: + data['message'] += converter.message + + if thread_ts: + data['root_id'] = thread_ts + + data['props']['attachments'] = [] + if converter.attachments_result: + data['props']['attachments'].extend(converter.attachments_result) + if attachments: + data['props']['attachments'].extend([a.to_dict() for a in attachments]) + + response = self.call_resource( + Resource('posts', 'POST'), json=data, + ) + return MattermostMessage( + self, response, text=text, raise_exc=raise_exc, blocks=blocks, attachments=attachments + ) + + def upload_file(self, + channel, file, *, + title: str = None, + content: str = None, + filename: str = None, + thread_ts: str = None, + filetype: str = 'text', + raise_exc: bool = False): + if not self._is_channel_id(channel): + channel = self.channel_id_by_name(channel) + + params = { + 'channel_id': channel, + } + + data = { + 'channel_id': channel + } + + if isinstance(file, str) and content: + params.update(filename=file) + data.update(content=content) + elif isinstance(file, str) and not content: + params.update(filename=os.path.basename(file)) + with open(file, 'r') as f: + data.update(content=f.read()) + else: + params.update(filename=filename or _random_string(7)) + data.update(content=file.read()) + + response = self.call_resource( + Resource('files', 'POST'), params=params, files=data, raise_exc=raise_exc + ) + file_id = response.json()['file_infos'][0]['id'] + data = { + 'file_ids': [file_id] + } + if thread_ts: + post_id = thread_ts + response = self.call_resource( + Resource(f'posts/{post_id}/patch', 'PUT'), + json=data, raise_exc=raise_exc, + ) + return MattermostMessage( + self, response, text='', raise_exc=raise_exc, blocks=[], attachments=[] + ) + + response = self.call_resource( + Resource('posts', 'POST'), + json=data, raise_exc=raise_exc, + ) + return MattermostMessage( + self, response, text='', raise_exc=raise_exc, blocks=[], attachments=[] + ) + + +def call_resource(*args, **kwargs): + return Mattermost.from_env().call_resource(*args, **kwargs) + + +def send_notify(*args, **kwargs): + return Mattermost.from_env().send_notify(*args, **kwargs) diff --git a/slack_notifications/mattermost/converter.py b/slack_notifications/mattermost/converter.py new file mode 100644 index 0000000..5bed0a3 --- /dev/null +++ b/slack_notifications/mattermost/converter.py @@ -0,0 +1,109 @@ +from typing import List + +from slack_notifications.fields import ( + BaseBlock, + HeaderBlock, + SimpleTextBlock, + SimpleTextBlockField, + DividerBlock, + ImageBlock, + ContextBlock, + ContextBlockTextElement, + ContextBlockImageElement, + ActionsBlock, + ButtonBlock, + Attachment +) + + +class MattermostConverter: + def __init__(self, *, header_level: int = 4, blocks: List[BaseBlock] = None): + self.message = '' + self._level = header_level + self._blocks = blocks or [] + self.attachments_result = [] + + self.block_handlers: dict[type[BaseBlock], callable] = { + HeaderBlock: self.convert_header_block, + SimpleTextBlock: self.convert_simple_text_block, + DividerBlock: self.convert_divider_block, + ImageBlock: self.convert_image_block, + ContextBlock: self.convert_context_block, + } + + def convert(self, blocks: List[BaseBlock]): + if not blocks: + return + + attachment = Attachment(color='gray') + attachment_data = attachment.to_dict() + attachment_data['actions'] = [] + + for block in blocks: + block_handler = self.block_handlers.get(type(block)) + if block_handler: + block_handler(block) + + if isinstance(block, ActionsBlock): + attachment_data['actions'] += self.convert_actions_block(block) + + if 'actions' in attachment_data: + + self.attachments_result.append(attachment_data) + + def fields_to_table(self, fields: List[SimpleTextBlockField]): + header = '| |' + separator = '|:---|---:|' + + table_rows = [] + for i in range(0, len(fields), 2): + row = fields[i:i + 2] + table_rows.append('|'+'|'.join(map(lambda v: str(v.text), row))+'|') + + return '\n'.join([header, separator] + table_rows) + + def convert_header_block(self, block: HeaderBlock): + sign = '#' + self.message += f'\n{sign * self._level} {block.text}' + + def convert_simple_text_block(self, block: SimpleTextBlock): + self.message += f'\n{block.text}\n\n{self.fields_to_table(block.fields)}\n'\ + if block.fields else f'\n{block.text}\n' + + def convert_divider_block(self, block: DividerBlock): + self.message += '\n---\n' + + def convert_image_block(self, block: ImageBlock): + self.message += f'\n![{block.title}]({block.image_url}\n)' + + def convert_context_block(self, block: ContextBlock): + self.message += '\n' + for element in block.elements: + if isinstance(element, ContextBlockTextElement): + self.message += f' *{element.text}* ' + if isinstance(element, ContextBlockImageElement): + self.message += f' ![{element.alt_text}]({element.image_url} =30) ' + self.message += '\n' + + def convert_actions_block(self, block: ActionsBlock): + data = [] + for element in block.elements: + if isinstance(element, ButtonBlock): + data.append(self.convert_button_block(element)) + return data + + def convert_button_block(self, block: ButtonBlock): + data = { + 'id': block.action_id, + 'type': block.__type__, + 'name': block.text, + 'integration': { + 'context': { + 'value': block.value + } + } + } + if block.style is not None: + data['style'] = block.style + + return data diff --git a/slack_notifications/mattermost/fields/__init__.py b/slack_notifications/mattermost/fields/__init__.py new file mode 100644 index 0000000..9fdd366 --- /dev/null +++ b/slack_notifications/mattermost/fields/__init__.py @@ -0,0 +1,5 @@ +from slack_notifications.mattermost.fields import mrkdwn + +__all__ = [ + 'mrkdwn', +] diff --git a/slack_notifications/mattermost/fields/mrkdwn.py b/slack_notifications/mattermost/fields/mrkdwn.py new file mode 100644 index 0000000..301209d --- /dev/null +++ b/slack_notifications/mattermost/fields/mrkdwn.py @@ -0,0 +1,41 @@ +from typing import Iterable + + +def bold(text): + return f'**{str(text)}**' + + +def italic(text): + return f'*{str(text)}*' + + +def strike(text): + return f'~~{str(text)}~~' + + +def code_block(text): + return f'`{str(text)}`' + + +def multi_line_code_block(text, *, lang=''): + return f'```{lang}\n{str(text)}\n```' + + +def block_quotes(text): + return f'>{str(text)}' + + +def list_items(itr: Iterable): + return ''.join([f'- {str(item)}' for item in itr]) + '\n' + + +def link(url, text): + return f'[{str(text)}]({url})' + + +def user_mention(user_id): + return f'@{user_id}' + + +def channel_mention(channel): + return f'~{channel}' diff --git a/slack_notifications/slack/__init__.py b/slack_notifications/slack/__init__.py new file mode 100644 index 0000000..020c36b --- /dev/null +++ b/slack_notifications/slack/__init__.py @@ -0,0 +1,17 @@ +from slack_notifications.slack.client import ( + Slack, + SlackMessage as Message, + call_resource, + send_notify, + resource_iterator, + ACCESS_TOKEN +) + +__all__ = [ + 'ACCESS_TOKEN', + 'Slack', + 'Message', + 'call_resource', + 'resource_iterator', + 'send_notify', +] \ No newline at end of file diff --git a/slack_notifications/slack/client.py b/slack_notifications/slack/client.py new file mode 100644 index 0000000..b1f16b3 --- /dev/null +++ b/slack_notifications/slack/client.py @@ -0,0 +1,246 @@ +import os +from typing import List +import requests +import logging + +from slack_notifications.common import Resource, NotificationClient +from slack_notifications.utils import _random_string +from slack_notifications.fields import BaseBlock, Attachment + +ACCESS_TOKEN = None +ACCESS_TOKEN_ENV_NAME = 'SLACK_ACCESS_TOKEN' + + +logger = logging.getLogger(__name__) + + +class SlackError(requests.exceptions.RequestException): + pass + + +class SlackMessage: + def __init__(self, client, response, + text: str = None, + raise_exc=False, + attachments: List[Attachment] = None, + blocks: List[BaseBlock] = None): + self._client = client + self._response = response + self._raise_exc = raise_exc + + self.text = text + self.attachments = attachments or [] + self.blocks = blocks or [] + + self.__lock_thread = False + + @property + def response(self): + return self._response + + def _lock_thread(self): + self.__lock_thread = True + + def send_to_thread(self, **kwargs): + if self.__lock_thread: + raise SlackError('Cannot open thread for thread message') + + json = self._response.json() + thread_ts = json['message']['ts'] + kwargs.update(thread_ts=thread_ts) + + message = self._client.send_notify(json['channel'], **kwargs) + + lock_thread = getattr(message, '_lock_thread') + lock_thread() + + return message + + def update(self): + json = self._response.json() + data = { + 'channel': json['channel'], + 'ts': json['message']['ts'], + } + + if self.text: + data['text'] = self.text + + if self.blocks: + data['blocks'] = [b.to_dict() for b in self.blocks] + + if self.attachments: + data['attachments'] = [a.to_dict() for a in self.attachments] + + return self._client.call_resource( + Resource('chat.update', 'POST'), + json=data, raise_exc=self._raise_exc, + ) + + def delete(self): + json = self._response.json() + data = { + 'channel': json['channel'], + 'ts': json['message']['ts'], + } + return self._client.call_resource( + Resource('chat.update', 'POST'), + json=data, raise_exc=self._raise_exc, + ) + + def upload_file(self, file, **kwargs): + json = self._response.json() + kwargs.update(thread_ts=json['message']['ts']) + return self._client.upload_file(json['channel'], file, **kwargs) + + def add_reaction(self, name, raise_exc=False): + json = self._response.json() + data = { + 'name': name, + 'channel': json['channel'], + 'timestamp': json['message']['ts'], + } + return self._client.call_resource( + Resource('reactions.add', 'POST'), + json=data, raise_exc=raise_exc, + ) + + def remove_reaction(self, name, raise_exc=False): + json = self._response.json() + data = { + 'name': name, + 'channel': json['channel'], + 'timestamp': json['message']['ts'], + } + return self._client.call_resource( + Resource('reactions.remove', 'POST'), + json=data, raise_exc=raise_exc, + ) + + +class Slack(NotificationClient): + API_URL = 'https://slack.com/api' + + def __init__(self, token: str): + super().__init__(self.API_URL, token=token) + + @classmethod + def from_env(cls): + token = ACCESS_TOKEN or os.getenv(ACCESS_TOKEN_ENV_NAME) + assert token is not None, 'Please export "{}" environment variable'.format(ACCESS_TOKEN_ENV_NAME) + return cls(token) + + def send_notify(self, + channel, *, + text: str = None, + username: str = None, + icon_url: str = None, + icon_emoji: str = None, + link_names: bool = True, + raise_exc: bool = False, + attachments: List[Attachment] = None, + blocks: List[BaseBlock] = None, + thread_ts: str = None) -> SlackMessage: + data = { + 'channel': channel, + 'link_names': link_names, + } + + if username: + data['username'] = username + + if text: + data['mrkdwn'] = True + data['text'] = text + + if icon_url: + data['icon_url'] = icon_url + + if icon_emoji: + data['icon_emoji'] = icon_emoji + + if blocks: + data['blocks'] = [b.to_dict() for b in blocks] + + if attachments: + data['attachments'] = [a.to_dict() for a in attachments] + + if thread_ts: + data['thread_ts'] = thread_ts + + response = self.call_resource( + Resource('chat.postMessage', 'POST'), raise_exc=raise_exc, json=data, + ) + return SlackMessage(self, response, text=text, raise_exc=raise_exc, blocks=blocks, attachments=attachments) + + def upload_file(self, + channel, file, *, + title: str = None, + content: str = None, + filename: str = None, + thread_ts: str = None, + filetype: str = 'text', + raise_exc: bool = False): + data = { + 'channels': channel, + 'filetype': filetype, + } + if isinstance(file, str) and content: + filename = file + data.update(content=content, filename=filename) + elif isinstance(file, str) and not content: + data.update(filename=os.path.basename(file)) + with open(file, 'r') as f: + data.update(content=f.read()) + else: + data.update(content=file.read(), filename=filename or _random_string(7)) + + if title: + data.update(title=title) + if thread_ts: + data.update(thread_ts=thread_ts) + + return self.call_resource( + Resource('files.upload', 'POST'), + data=data, + raise_exc=raise_exc, + headers={ + 'Content-Type': 'application/x-www-form-urlencoded', + }, + ) + + def resource_iterator(self, + resource: Resource, from_key: str, *, + cursor: str = None, + raise_exc: bool = False, + limit: int = None): + params = {'limit': limit} + + if cursor: + params['cursor'] = cursor + + response = self.call_resource(resource, params=params, raise_exc=raise_exc) + data = response.json() + + for item in data[from_key]: + yield item + + cursor = data.get('response_metadata', {}).get('next_cursor') + + if cursor: + yield from self.resource_iterator( + resource, from_key, + limit=limit or self.DEFAULT_RECORDS_LIMIT, cursor=cursor, raise_exc=raise_exc, + ) + + +def call_resource(*args, **kwargs): + return Slack.from_env().call_resource(*args, **kwargs) + + +def resource_iterator(*args, **kwargs): + return Slack.from_env().resource_iterator(*args, **kwargs) + + +def send_notify(*args, **kwargs): + return Slack.from_env().send_notify(*args, **kwargs) diff --git a/slack_notifications/utils.py b/slack_notifications/utils.py new file mode 100644 index 0000000..d80386b --- /dev/null +++ b/slack_notifications/utils.py @@ -0,0 +1,13 @@ +import string +import random + +from slack_notifications.constants import COLOR_MAP + + +def _random_string(length): + letters = string.ascii_lowercase + return ''.join(random.choice(letters) for i in range(length)) + + +def init_color(name, code): + COLOR_MAP[name] = code