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
|
import warnings
from six import BytesIO, text_type
from six.moves.urllib.parse import urlparse, parse_qsl
from .util import CaseInsensitiveDict
import logging
log = logging.getLogger(__name__)
class Request(object):
"""
VCR's representation of a request.
"""
def __init__(self, method, uri, body, headers):
self.method = method
self.uri = uri
self._was_file = hasattr(body, "read")
if self._was_file:
self.body = body.read()
else:
self.body = body
self.headers = headers
log.debug("Invoking Request %s", self.uri)
@property
def headers(self):
return self._headers
@headers.setter
def headers(self, value):
if not isinstance(value, HeadersDict):
value = HeadersDict(value)
self._headers = value
@property
def body(self):
return BytesIO(self._body) if self._was_file else self._body
@body.setter
def body(self, value):
if isinstance(value, text_type):
value = value.encode("utf-8")
self._body = value
def add_header(self, key, value):
warnings.warn(
"Request.add_header is deprecated. " "Please assign to request.headers instead.",
DeprecationWarning,
)
self.headers[key] = value
@property
def scheme(self):
return urlparse(self.uri).scheme
@property
def host(self):
return urlparse(self.uri).hostname
@property
def port(self):
parse_uri = urlparse(self.uri)
port = parse_uri.port
if port is None:
try:
port = {"https": 443, "http": 80}[parse_uri.scheme]
except KeyError:
pass
return port
@property
def path(self):
return urlparse(self.uri).path
@property
def query(self):
q = urlparse(self.uri).query
return sorted(parse_qsl(q))
# alias for backwards compatibility
@property
def url(self):
return self.uri
# alias for backwards compatibility
@property
def protocol(self):
return self.scheme
def __str__(self):
return "<Request ({}) {}>".format(self.method, self.uri)
def __repr__(self):
return self.__str__()
def _to_dict(self):
return {
"method": self.method,
"uri": self.uri,
"body": self.body,
"headers": {k: [v] for k, v in self.headers.items()},
}
@classmethod
def _from_dict(cls, dct):
return Request(**dct)
class HeadersDict(CaseInsensitiveDict):
"""
There is a weird quirk in HTTP. You can send the same header twice. For
this reason, headers are represented by a dict, with lists as the values.
However, it appears that HTTPlib is completely incapable of sending the
same header twice. This puts me in a weird position: I want to be able to
accurately represent HTTP headers in cassettes, but I don't want the extra
step of always having to do [0] in the general case, i.e.
request.headers['key'][0]
In addition, some servers sometimes send the same header more than once,
and httplib *can* deal with this situation.
Furthermore, I wanted to keep the request and response cassette format as
similar as possible.
For this reason, in cassettes I keep a dict with lists as keys, but once
deserialized into VCR, I keep them as plain, naked dicts.
"""
def __setitem__(self, key, value):
if isinstance(value, (tuple, list)):
value = value[0]
# Preserve the case from the first time this key was set.
old = self._store.get(key.lower())
if old:
key = old[0]
super(HeadersDict, self).__setitem__(key, value)
|
