/
client.py
124 lines (105 loc) · 4.55 KB
/
client.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
# localslackirc
# Copyright (C) 2018 Salvo "LtWorf" Tomaselli
#
# localslackirc is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# author Salvo "LtWorf" Tomaselli <tiposchi@tiscali.it>
#
# This file was part of python-slackclient
# (https://github.com/slackapi/python-slackclient)
# But has been copied and relicensed under GPL. The copyright applies only
# to the changes made since it was copied.
import json
import logging
from typing import Any, Dict, List, Optional
from .server import Server, LoginInfo
from .exceptions import *
LOG = logging.getLogger(__name__)
class SlackClient:
'''
The SlackClient makes API Calls to the `Slack Web API <https://api.slack.com/web>`_ as well as
managing connections to the `Real-time Messaging API via websocket <https://api.slack.com/rtm>`_
It also manages some of the Client state for Channels that the associated token (User or Bot)
is associated with.
For more information, check out the `Slack API Docs <https://api.slack.com/>`_
Init:
:Args:
token (str): Your Slack Authentication token. You can find or generate a test token
`here <https://api.slack.com/docs/oauth-test-tokens>`_
Note: Be `careful with your token <https://api.slack.com/docs/oauth-safety>`_
proxies (dict): Proxies to use when create websocket or api calls,
declare http and websocket proxies using {'http': 'http://127.0.0.1'},
and https proxy using {'https': 'https://127.0.0.1:443'}
'''
def __init__(self, token: str, proxies: Optional[Dict[str,str]] = None) -> None:
self.token = token
self.server = Server(self.token, proxies)
def rtm_connect(self) -> LoginInfo:
'''
Connects to the RTM Websocket
:Returns:
False on exceptions
'''
self.server.rtm_connect()
if self.server.login_data is None:
raise SlackConnectionError('No login data available')
return self.server.login_data
@property
def fileno(self) -> Optional[int]:
return self.server.ws_fileno
def api_call(self, method: str, timeout: Optional[float] = None, **kwargs) -> Dict[str, Any]:
'''
Call the Slack Web API as documented here: https://api.slack.com/web
:Args:
method (str): The API Method to call. See
`the full list here <https://api.slack.com/methods>`_
:Kwargs:
(optional) kwargs: any arguments passed here will be bundled and sent to the api
requester as post_data and will be passed along to the API.
Example::
sc.server.api_call(
"channels.setPurpose",
channel="CABC12345",
purpose="Writing some code!"
)
'''
response_body = self.server.api_call(method, timeout=timeout, **kwargs)
try:
result = response_body
except ValueError as json_decode_error:
raise ParseResponseError(response_body, json_decode_error)
return result
def rtm_read(self) -> List[Dict[str, Any]]:
'''
Reads from the RTM Websocket stream then calls `self.process_changes(item)` for each line
in the returned data.
Multiple events may be returned, always returns a list [], which is empty if there are no
incoming messages.
:Returns:
data (json) - The server response. For example::
[{u'presence': u'active', u'type': u'presence_change', u'user': u'UABC1234'}]
:Raises:
SlackNotConnected if self.server is not defined.
'''
# in the future, this should handle some events internally i.e. channel
# creation
if self.server:
json_data = self.server.websocket_read()
data = []
if json_data != '':
for d in json_data.split('\n'):
data.append(json.loads(d))
return data
else:
raise SlackNotConnected