-
Notifications
You must be signed in to change notification settings - Fork 200
/
utils.py
116 lines (96 loc) · 3.25 KB
/
utils.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
"""
This module contains common utilities used by the connector
"""
from typing import Any, Dict, Optional
import http.client
import urllib.parse
class Request:
"""
Provides a wrapper for the python http.client,
to be used similar to the requests library.
Parameters
----------
_url: The requesting end-point URL.
"""
def __init__(self, _url: str) -> None:
self.url: urllib.parse.ParseResult = urllib.parse.urlparse(_url)
qs = dict(urllib.parse.parse_qsl(self.url.query))
self.url = self.url._replace(query=urllib.parse.urlencode(qs))
self.hostname: str = self.url.hostname or ""
self.headers: Dict[str, Any] = dict({"user-agent": "dataprep"})
def get(self, _headers: Optional[Dict[str, Any]] = None) -> http.client.HTTPResponse:
"""
GET request to the specified end-point.
Parameters
----------
_headers: Any additional headers to be passed
"""
if _headers:
self.headers.update(_headers)
conn = http.client.HTTPSConnection(self.hostname)
path = self.url.path
if path == "":
path = "/"
if self.url.query:
path += "?" + self.url.query
conn.request(method="GET", url=path, headers=self.headers)
response = conn.getresponse()
return response
def post(
self, _headers: Optional[Dict[str, Any]] = None, _data: Optional[Dict[str, Any]] = None
) -> http.client.HTTPResponse:
"""
POST request to the specified end-point.
Parameters
----------
_headers: Any additional headers to be passed
_data: Body of the request
"""
if _headers:
self.headers.update(_headers)
conn = http.client.HTTPSConnection(self.hostname)
path = self.url.path
if path == "":
path = "/"
if self.url.query:
path += "?" + self.url.query
if _data is not None:
conn.request(
method="POST",
url=path,
headers=self.headers,
body=urllib.parse.urlencode(_data),
)
else:
conn.request(method="POST", url=path, headers=self.headers)
response = conn.getresponse()
return response
def put(
self, _headers: Optional[Dict[str, Any]] = None, _data: Optional[Dict[str, Any]] = None
) -> http.client.HTTPResponse:
"""
PUT request to the specified end-point.
Parameters
----------
_headers: Any additional headers to be passed
_data: Body of the request
"""
if _headers:
self.headers.update(_headers)
conn = http.client.HTTPSConnection(self.hostname)
path = self.url.path
if path == "":
path = "/"
if self.url.query:
path += "?" + self.url.query
if _data is not None:
conn.request(
method="PUT",
url=path,
headers=self.headers,
body=urllib.parse.urlencode(_data),
)
else:
conn.request(method="PUT", url=path, headers=self.headers)
response = conn.getresponse()
return response