forked from pubnub/python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathstructures.py
More file actions
96 lines (77 loc) · 3.03 KB
/
structures.py
File metadata and controls
96 lines (77 loc) · 3.03 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
from .enums import HttpMethod
class RequestOptions(object):
def __init__(
self, path, params_callback,
method, request_timeout, connect_timeout,
create_response, create_status, create_exception,
operation_type, data=None, sort_arguments=False,
allow_redirects=True, use_base_path=None, files=None,
request_headers=None, non_json_response=False
):
assert len(path) > 0
assert callable(params_callback)
assert isinstance(method, int)
assert isinstance(request_timeout, int)
assert isinstance(connect_timeout, int)
if not (method is HttpMethod.GET or method is HttpMethod.POST or method is HttpMethod.DELETE
or method is HttpMethod.PATCH): # noqa
raise AssertionError()
self.params = None
self.path = path
self.params_callback = params_callback
self._method = method
self.request_timeout = request_timeout
self.connect_timeout = connect_timeout
# TODO: rename 'data' => 'body'
self.data = data
self.files = files
self.body = data
self.sort_params = sort_arguments
self.create_response = create_response
self.create_status = create_status
self.create_exception = create_exception
self.operation_type = operation_type
self.allow_redirects = allow_redirects
self.use_base_path = use_base_path
self.request_headers = request_headers
self.non_json_response = non_json_response
def merge_params_in(self, params_to_merge_in):
self.params = self.params_callback(params_to_merge_in)
@property
def method_string(self):
return HttpMethod.string(self._method)
def is_post(self):
return self._method is HttpMethod.POST
def is_patch(self):
return self._method is HttpMethod.PATCH
def query_list(self):
""" All query keys and values should be already encoded inside a build_params() method"""
s = []
for k, v in self.params.items():
s.append(str(k) + "=" + str(v))
if self.sort_params:
return sorted(s)
else:
return s
@property
def query_string(self):
return str('&'.join(self.query_list()))
def __str__(self):
return "path: {0}, qs: {1}".format(self.path, self.query_string)
class PlatformOptions(object):
def __init__(self, headers, pn_config):
self.headers = headers
self.pn_config = pn_config
class ResponseInfo(object):
def __init__(self, status_code, tls_enabled, origin, uuid, auth_key, client_request, client_response=None):
self.status_code = status_code
self.tls_enabled = tls_enabled
self.origin = origin
self.uuid = uuid
self.auth_key = auth_key
self.client_request = client_request
self.client_response = client_response
class Envelope(object):
def __init__(self, result, status):
self.result = result
self.status = status