Source code for parsons.notifications.slack

from parsons import Table
from slackclient import SlackClient
from slackclient.exceptions import SlackClientError
import os
import time


[docs]class Slack(object): def __init__(self, api_key=None): if api_key is None: try: self.api_key = os.environ["SLACK_API_TOKEN"] except KeyError: raise KeyError('Missing api_key. It must be passed as an ' 'argument or stored as environmental variable') else: self.api_key = api_key self.client = SlackClient(self.api_key)
[docs] def channels(self, fields=['id', 'name'], exclude_archived=False, types=['public_channel']): """ Return a list of all channels in a Slack team. `Args:` fields: list A list of the fields to return. By default, only the channel `id` and `name` are returned. See https://api.slack.com/methods/conversations.list for a full list of available fields. `Notes:` nested fields are unpacked. exclude_archived: bool Set to `True` to exclude archived channels from the list. Default is false. types: list Mix and match channel types by providing a list of any combination of `public_channel`, `private_channel`, `mpim` (aka group messages), or `im` (aka 1-1 messages). `Returns:` Parsons Table See :ref:`parsons-table` for output options. """ tbl = self._paginate_request( "conversations.list", "channels", types=types, exclude_archived=exclude_archived) tbl.unpack_dict("topic", include_original=False, prepend=True, prepend_value="topic") tbl.unpack_dict("purpose", include_original=False, prepend=True, prepend_value="purpose") rm_cols = [x for x in tbl.columns if x not in fields] tbl.remove_column(*rm_cols) return tbl
[docs] def users(self, fields=['id', 'name', 'deleted', 'profile_real_name_normalized', 'profile_email']): """ Return a list of all users in a Slack team. `Args:` fields: list A list of the fields to return. By default, only the user `id` and `name` and `deleted` status are returned. See https://api.slack.com/methods/users.list for a full list of available fields. `Notes:` nested fields are unpacked. `Returns:` Parsons Table See :ref:`parsons-table` for output options. """ tbl = self._paginate_request("users.list", "members", include_locale=True) tbl.unpack_dict("profile", include_original=False, prepend=True, prepend_value="profile") rm_cols = [x for x in tbl.columns if x not in fields] tbl.remove_column(*rm_cols) return tbl
[docs] def message_channel(self, channel, text, as_user=False): """ Send a message to a Slack channel `Args:` channel: str The name or id of a `public_channel`, a `private_channel`, or an `im` (aka 1-1 message). text: str Text of the message to send. as_user: str Pass true to post the message as the authenticated user, instead of as a bot. Defaults to false. See https://api.slack.com/methods/chat.postMessage#authorship for more information about Slack authorship. `Returns:` `dict`: A response json """ resp = self.client.api_call( "chat.postMessage", channel=channel, text=text, as_user=as_user) if not resp['ok']: if resp['error'] == 'ratelimited': time.sleep(int(resp['headers']['Retry-After'])) resp = self.client.api_call( "chat.postMessage", channel=channel, text=text, as_user=as_user) raise SlackClientError(resp['error']) return resp
[docs] def upload_file(self, channels, filename, filetype=None, initial_comment=None, title=None, is_binary=False): """ Upload a file to Slack channel(s). `Args:` channels: list The list of channel names or IDs where the file will be shared. filename: str The path to the file to be uploaded. filetype: str A file type identifier. If None, type will be inferred base on file extension. This is used to determine what fields are available for that object. See https://api.slack.com/types/file for a list of valid types and for more information about the file object. initial_comment: str The text of the message to send along with the file. title: str Title of the file to be uploaded. is_binary: bool If True, open this file in binary mode. This is needed if uploading binary files. Defaults to False. `Returns:` `dict`: A response json """ if filetype is None and '.' in filename: filetype = filename.split('.')[-1] mode = 'rb' if is_binary else 'r' with open(filename, mode) as file_content: resp = self.client.api_call( "files.upload", channels=channels, file=file_content, filetype=filetype, initial_comment=initial_comment, title=title) if not resp['ok']: if resp['error'] == 'ratelimited': time.sleep(int(resp['headers']['Retry-After'])) resp = self.client.api_call( "files.upload", channels=channels, file=file_content, filetype=filetype, initial_comment=initial_comment, title=title) raise SlackClientError(resp['error']) return resp
def _paginate_request(self, endpoint, collection, **kwargs): # The max object we're requesting at a time. # This is an nternal limit to not overload slack api LIMIT = 200 items = [] next_page = True cursor = None while next_page: resp = self.client.api_call( endpoint, cursor=cursor, limit=LIMIT, **kwargs) if not resp['ok']: if resp['error'] == 'ratelimited': time.sleep(int(resp['headers']['Retry-After'])) continue raise SlackClientError(resp['error']) items.extend(resp[collection]) if resp["response_metadata"]["next_cursor"]: cursor = resp["response_metadata"]["next_cursor"] else: next_page = False return Table(items)