-
Notifications
You must be signed in to change notification settings - Fork 0
/
requests2.py
140 lines (119 loc) · 3.85 KB
/
requests2.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
import usocket
#import network
#esp = network.WLAN(network.STA_IF)
#esp.active(True)
#esp.connect('madagscar','lillianEMMA')
class Response:
def __init__(self, f):
self.raw = f
self.encoding = "utf-8"
self._cached = None
def close(self):
if self.raw:
self.raw.close()
self.raw = None
self._cached = None
@property
def content(self):
if self._cached is None:
self._cached = self.raw.read()
self.raw.close()
self.raw = None
return self._cached
@property
def text(self):
return str(self.content, self.encoding)
def json(self):
import ujson
return ujson.loads(self.content)
class Http:
def request(self, method, url, data=None, json=None, headers={}, stream=None):
try:
proto, dummy, host, path = url.split("/", 3)
except ValueError:
proto, dummy, host = url.split("/", 2)
path = ""
if proto == "http:":
port = 80
elif proto == "https:":
import ussl
port = 443
else:
raise ValueError("Unsupported protocol: " + proto)
if ":" in host:
host, port = host.split(":", 1)
port = int(port)
# print("addres {0}:{1} | proto {2} | path {3}".format(host, port, proto, path))
ai = usocket.getaddrinfo(host, port)
# print(ai)
addr = ai[0][-1]
s = usocket.socket()
# print(addr)
s.connect(addr)
if proto == "https:":
s = ussl.wrap_socket(s)
s.write(b"%s /%s HTTP/1.0\r\n" % (method, path))
if not "Host" in headers:
s.write(b"Host: %s\r\n" % host)
# Iterate over keys to avoid tuple alloc
for k in headers:
s.write(k)
s.write(b": ")
s.write(headers[k])
s.write(b"\r\n")
if json is not None:
assert data is None
import ujson
data = ujson.dumps(json)
if data:
s.write(b"Content-Length: %d\r\n" % len(data))
s.write(b"\r\n")
# print(str(data))
if data:
s.write(str(data))
l = s.readline()
protover, status, msg = l.split(None, 2)
status = int(status)
#print(protover, status, msg)
# Added redirect support
if status in range(300, 309):
return self.request(method, url, data, json, headers, stream)
while True:
l = s.readline()
if not l or l == b"\r\n":
break
#print(line)
if l.startswith(b"Transfer-Encoding:"):
if b"chunked" in line:
raise ValueError("Unsupported " + l)
elif l.startswith(b"Location:"):
raise NotImplementedError("Redirects not yet supported")
resp = Response(s)
resp.status_code = status
resp.reason = msg.rstrip()
return resp
# @classmethod
def head(cls, url, **kw):
return cls.request("HEAD", url, **kw)
# @classmethod
def get(cls, url, **kw):
return cls.request("GET", url, **kw)
# @classmethod
def post(cls, url, **kw):
return cls.request("POST", url, **kw)
# @classmethod
def put(url, **kw):
return Http.request("PUT", url, **kw)
# @classmethod
def patch(url, **kw):
return Http.request("PATCH", url, **kw)
# @classmethod
def delete(url, **kw):
return Http.request("DELETE", url, **kw)
@classmethod
def test():
r = (post('http://192.168.1.76:3000/sensors/1/readings.json',
json={'reading': {'value': 99}},
headers={'Content-Type':'application/json'}))
# print(r.status_code)
return r