-
Notifications
You must be signed in to change notification settings - Fork 95
Expand file tree
/
Copy pathresponse.py
More file actions
69 lines (54 loc) · 1.97 KB
/
response.py
File metadata and controls
69 lines (54 loc) · 1.97 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
import requests
from v20.errors import ResponseUnexpectedStatus, ResponseNoField, V20Timeout, V20ConnectionError
class Response(object):
def __init__(self, request, method, path, status, reason, headers):
self.request = request
self.method = method
self.path = path
self.status = status
self.reason = reason
self.headers = headers
self.content_type = headers.get("content-type", None)
self.raw_body = None
self.body = None
self.lines = None
self.line_parser = None
def set_raw_body(self, raw_body):
self.raw_body = raw_body
def set_lines(self, lines):
self.lines = lines
def set_line_parser(self, parser):
self.line_parser = parser
def get(self, field, status=None):
if status is not None:
if str(self.status) != str(status):
raise ResponseUnexpectedStatus(self, status)
value = self.body.get(field)
if value is None:
raise ResponseNoField(self, field)
return value
def parts(self):
def line_parser(line):
return "line", line
parser = line_parser
if self.line_parser is not None:
parser = self.line_parser
if self.lines is None:
return
try:
for line in self.lines:
yield parser(line)
except requests.exceptions.ConnectionError:
raise V20Timeout(self.path, "stream")
except requests.exceptions.ChunkedEncodingError:
raise V20ConnectionError(self.path)
def __str__(self):
s = "Method = {}\n".format(self.method)
s += "Path = {}\n".format(self.path)
s += "Status = {}\n".format(self.status)
s += "Reason = {}\n".format(self.reason)
s += "Content-Type = {}\n".format(self.content_type)
s += "Raw_Body = {}\n".format(self.raw_body)
return s
def __repr__(self):
return str(self)