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
|
# Copyright (C) 2018 Lars Wirzenius
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import json
import requests
class HttpError(Exception):
pass
class HttpAPI:
# Make requests to an HTTP API.
json_type = 'application/json'
def __init__(self):
self._session = requests.Session()
self._token = None
def set_session(self, session):
self._session = session
def set_token(self, token):
self._token = token
def get_dict(self, url, headers=None):
r = self._request(self._session.get, url, headers=headers)
ct = r.headers.get('Content-Type')
if ct != self.json_type:
raise HttpError('Not JSON response')
try:
return r.json()
except json.decoder.JSONDecodeError:
raise HttpError('JSON parsing error')
def get_blob(self, url, headers=None):
r = self._request(self._session.get, url, headers=headers)
return r.content
def post(self, url, headers=None, body=None):
self._send_request(self._session.post, url, headers=headers, body=body)
return None
def put(self, url, headers=None, body=None):
self._send_request(self._session.put, url, headers=headers, body=body)
return None
def _send_request(self, func, url, headers=None, body=None):
if headers is None:
headers = {}
headers = dict(headers)
h, body = self._get_content_type_header(body)
headers.update(h)
self._request(func, url, headers=headers, data=body)
return None
def _get_content_type_header(self, body):
if isinstance(body, dict):
header = {
'Content-Type': 'application/json',
}
body = json.dumps(body)
return header, body
return {}, body
def _get_authorization_headers(self):
return {
'Authorization': 'Bearer {}'.format(self._token),
}
def _request(self, func, url, headers=None, **kwargs):
if headers is None:
headers = {}
headers.update(self._get_authorization_headers())
logging.debug('request: func=%r', func)
logging.debug('request: url=%r', url)
for h in headers:
logging.debug('request: %s: %s', h, headers[h])
logging.debug('request: kwargs=%r', kwargs)
r = func(url, headers=headers, verify=False, **kwargs)
logging.debug('response: status_code=%r', r.status_code)
logging.debug('response: content=%r', r.content)
if not r.ok:
raise HttpError(r.status_code)
return r
|
