-
Notifications
You must be signed in to change notification settings - Fork 9
/
url.py
154 lines (118 loc) · 4.78 KB
/
url.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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
"""
Convenient way to request HTTP APIs.
For examples, see http://github.com/kimmobrunfeldt/nap
"""
import sys
_PY3 = sys.version_info >= (3, 0)
# For Python 3 compatibility
if _PY3:
from urllib.parse import urljoin
else:
from urlparse import urljoin
import requests
class Url(object):
"""Wrapper class for requests library."""
def __init__(self, base_url, **default_kwargs):
"""
* `base_url`
API's base url. Trailing slash is optional.
For example `'https://api.github.com'`
* `**default_kwargs`
Keyword arguments that will be passed to
`requests.request` on each request
"""
self._base_url = base_url
self._default_kwargs = default_kwargs
@property
def url(self):
"""Returns base url"""
return self._base_url
def join(self, relative_url):
"""Joins base url with relative_url and returns new Url object
from the combined url.
"""
return self._new_url(relative_url)
# HTTP methods
def delete(self, *args, **kwargs):
return self._request('DELETE', *args, **kwargs)
def get(self, *args, **kwargs):
return self._request('GET', *args, **kwargs)
def head(self, *args, **kwargs):
return self._request('HEAD', *args, **kwargs)
def patch(self, *args, **kwargs):
return self._request('PATCH', *args, **kwargs)
def post(self, *args, **kwargs):
return self._request('POST', *args, **kwargs)
def put(self, *args, **kwargs):
return self._request('PUT', *args, **kwargs)
# Overridable methods to extend behavior
def before_request(self, method, relative_url, request_kwargs):
"""This method can be overridden to customize each request.
* `method`
The HTTP method of request in upper case. For example `'GET'`.
* `relative_url`
The relative URL passed to the HTTP method, without leading slash.
* `request_kwargs`
Keyword arguments that were passed to the request method.
This does not contain the default keyword arguments given when
initializing Api class. For example in `url.get(verify=False)`,
the value would be `{'verify': False}`.
This method should return keyword arguments. These returned kwargs
will be added on top of default request_kwargs given to class.
The sum of both keyword arguments will be passed to `requests.request`
"""
return request_kwargs
def after_request(self, response):
"""This method can be overridden to add default behavior when response
is returned. For example if you're working with a JSON API, you can
return deserialized JSON from this method instead of `Response` object.
* `response`
`requests.Response` object returned by *request* function.
The returned value will be returned to the API method caller:
`response = api.get('users')`.
"""
return response
def default_kwargs(self):
"""This method can be overridden to modify `default_kwargs` given in
class initialization.
Returns new default kwargs.
"""
return self._default_kwargs
def _request(self, http_method, relative_url='', **kwargs):
"""Does actual HTTP request using requests library."""
# It could be possible to call api.resource.get('/index')
# but it would be non-intuitive that the path would resolve
# to root of domain
relative_url = self._remove_leading_slash(relative_url)
# Add default kwargs with possible custom kwargs returned by
# before_request
new_kwargs = self.default_kwargs().copy()
custom_kwargs = self.before_request(
http_method,
relative_url,
kwargs.copy()
)
new_kwargs.update(custom_kwargs)
response = requests.request(
http_method,
self._join_url(relative_url),
**new_kwargs
)
return self.after_request(response)
def _join_url(self, relative_url):
"""Joins relative url with base url. Adds trailing slash if needed."""
joined_url = urljoin(self._base_url, relative_url)
return joined_url
def _remove_leading_slash(self, text):
return text[1:] if text.startswith('/') else text
def _ensure_trailing_slash(self, text):
return text if text.endswith('/') else text + '/'
def _new_url(self, relative_url):
"""Create new Url which points to new url."""
return Url(
urljoin(
self._ensure_trailing_slash(self._base_url),
self._remove_leading_slash(relative_url)
),
**self._default_kwargs
)