forked from slackapi/python-slack-sdk
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathslackrequest.py
More file actions
105 lines (85 loc) · 4 KB
/
Copy pathslackrequest.py
File metadata and controls
105 lines (85 loc) · 4 KB
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
import requests
import json
import six
import sys
import platform
from .version import __version__
class SlackRequest(object):
def __init__(self, proxies=None):
# __name__ returns 'slackclient.slackrequest', we only want 'slackclient'
client_name = __name__.split('.')[0]
client_version = __version__ # Version is returned from version.py
# Construct the user-agent header with the package info, Python version and OS version.
self.default_user_agent = {
"client": "{0}/{1}".format(client_name, client_version),
"python": "Python/{v.major}.{v.minor}.{v.micro}".format(v=sys.version_info),
"system": "{0}/{1}".format(platform.system(), platform.release())
}
self.custom_user_agent = None
self.proxies = proxies
def get_user_agent(self):
# Check for custom user-agent and append if found
if self.custom_user_agent:
custom_ua_list = ["/".join(client_info) for client_info in self.custom_user_agent]
custom_ua_string = " ".join(custom_ua_list)
self.default_user_agent['custom'] = custom_ua_string
# Concatenate and format the user-agent string to be passed into request headers
ua_string = []
for key, val in self.default_user_agent.items():
ua_string.append(val)
user_agent_string = " ".join(ua_string)
return user_agent_string
def append_user_agent(self, name, version):
if self.custom_user_agent:
self.custom_user_agent.append([name.replace("/", ":"), version.replace("/", ":")])
else:
self.custom_user_agent = [[name, version]]
def do(self, token, request="?", post_data=None, domain="slack.com", timeout=None):
"""
Perform a POST request to the Slack Web API
Args:
token (str): your authentication token
request (str): the method to call from the Slack API. For example: 'channels.list'
timeout (float): stop waiting for a response after a given number of seconds
post_data (dict): key/value arguments to pass for the request. For example:
{'channel': 'CABC12345'}
domain (str): if for some reason you want to send your request to something other
than slack.com
"""
url = 'https://{0}/api/{1}'.format(domain, request)
# Override token header if `token` is passed in post_data
if post_data is not None and "token" in post_data:
token = post_data['token']
# Set user-agent and auth headers
headers = {
'user-agent': self.get_user_agent(),
'Authorization': 'Bearer {}'.format(token)
}
# Pull file out so it isn't JSON encoded like normal fields.
# Only do this for requests that are UPLOADING files; downloading files
# use the 'file' argument to point to a File ID.
post_data = post_data or {}
# Move singular file objects into `files`
upload_requests = ['files.upload']
# Move file content into requests' `files` param
files = None
if request in upload_requests:
files = {'file': post_data.pop('file')} if 'file' in post_data else None
# Check for plural fields and convert them to comma-separated strings if needed
for field in {'channels', 'users', 'types'} & set(post_data.keys()):
if isinstance(post_data[field], list):
post_data[field] = ",".join(post_data[field])
# Convert any params which are list-like to JSON strings
# Example: `attachments` is a dict, and needs to be passed as JSON
for k, v in six.iteritems(post_data):
if isinstance(v, (list, dict)):
post_data[k] = json.dumps(v)
# Submit the request
return requests.post(
url,
headers=headers,
data=post_data,
files=files,
timeout=timeout,
proxies=self.proxies
)