jpayne@7
|
1 from __future__ import annotations
|
jpayne@7
|
2
|
jpayne@7
|
3 import json as _json
|
jpayne@7
|
4 import typing
|
jpayne@7
|
5 from urllib.parse import urlencode
|
jpayne@7
|
6
|
jpayne@7
|
7 from ._base_connection import _TYPE_BODY
|
jpayne@7
|
8 from ._collections import HTTPHeaderDict
|
jpayne@7
|
9 from .filepost import _TYPE_FIELDS, encode_multipart_formdata
|
jpayne@7
|
10 from .response import BaseHTTPResponse
|
jpayne@7
|
11
|
jpayne@7
|
12 __all__ = ["RequestMethods"]
|
jpayne@7
|
13
|
jpayne@7
|
14 _TYPE_ENCODE_URL_FIELDS = typing.Union[
|
jpayne@7
|
15 typing.Sequence[typing.Tuple[str, typing.Union[str, bytes]]],
|
jpayne@7
|
16 typing.Mapping[str, typing.Union[str, bytes]],
|
jpayne@7
|
17 ]
|
jpayne@7
|
18
|
jpayne@7
|
19
|
jpayne@7
|
20 class RequestMethods:
|
jpayne@7
|
21 """
|
jpayne@7
|
22 Convenience mixin for classes who implement a :meth:`urlopen` method, such
|
jpayne@7
|
23 as :class:`urllib3.HTTPConnectionPool` and
|
jpayne@7
|
24 :class:`urllib3.PoolManager`.
|
jpayne@7
|
25
|
jpayne@7
|
26 Provides behavior for making common types of HTTP request methods and
|
jpayne@7
|
27 decides which type of request field encoding to use.
|
jpayne@7
|
28
|
jpayne@7
|
29 Specifically,
|
jpayne@7
|
30
|
jpayne@7
|
31 :meth:`.request_encode_url` is for sending requests whose fields are
|
jpayne@7
|
32 encoded in the URL (such as GET, HEAD, DELETE).
|
jpayne@7
|
33
|
jpayne@7
|
34 :meth:`.request_encode_body` is for sending requests whose fields are
|
jpayne@7
|
35 encoded in the *body* of the request using multipart or www-form-urlencoded
|
jpayne@7
|
36 (such as for POST, PUT, PATCH).
|
jpayne@7
|
37
|
jpayne@7
|
38 :meth:`.request` is for making any kind of request, it will look up the
|
jpayne@7
|
39 appropriate encoding format and use one of the above two methods to make
|
jpayne@7
|
40 the request.
|
jpayne@7
|
41
|
jpayne@7
|
42 Initializer parameters:
|
jpayne@7
|
43
|
jpayne@7
|
44 :param headers:
|
jpayne@7
|
45 Headers to include with all requests, unless other headers are given
|
jpayne@7
|
46 explicitly.
|
jpayne@7
|
47 """
|
jpayne@7
|
48
|
jpayne@7
|
49 _encode_url_methods = {"DELETE", "GET", "HEAD", "OPTIONS"}
|
jpayne@7
|
50
|
jpayne@7
|
51 def __init__(self, headers: typing.Mapping[str, str] | None = None) -> None:
|
jpayne@7
|
52 self.headers = headers or {}
|
jpayne@7
|
53
|
jpayne@7
|
54 def urlopen(
|
jpayne@7
|
55 self,
|
jpayne@7
|
56 method: str,
|
jpayne@7
|
57 url: str,
|
jpayne@7
|
58 body: _TYPE_BODY | None = None,
|
jpayne@7
|
59 headers: typing.Mapping[str, str] | None = None,
|
jpayne@7
|
60 encode_multipart: bool = True,
|
jpayne@7
|
61 multipart_boundary: str | None = None,
|
jpayne@7
|
62 **kw: typing.Any,
|
jpayne@7
|
63 ) -> BaseHTTPResponse: # Abstract
|
jpayne@7
|
64 raise NotImplementedError(
|
jpayne@7
|
65 "Classes extending RequestMethods must implement "
|
jpayne@7
|
66 "their own ``urlopen`` method."
|
jpayne@7
|
67 )
|
jpayne@7
|
68
|
jpayne@7
|
69 def request(
|
jpayne@7
|
70 self,
|
jpayne@7
|
71 method: str,
|
jpayne@7
|
72 url: str,
|
jpayne@7
|
73 body: _TYPE_BODY | None = None,
|
jpayne@7
|
74 fields: _TYPE_FIELDS | None = None,
|
jpayne@7
|
75 headers: typing.Mapping[str, str] | None = None,
|
jpayne@7
|
76 json: typing.Any | None = None,
|
jpayne@7
|
77 **urlopen_kw: typing.Any,
|
jpayne@7
|
78 ) -> BaseHTTPResponse:
|
jpayne@7
|
79 """
|
jpayne@7
|
80 Make a request using :meth:`urlopen` with the appropriate encoding of
|
jpayne@7
|
81 ``fields`` based on the ``method`` used.
|
jpayne@7
|
82
|
jpayne@7
|
83 This is a convenience method that requires the least amount of manual
|
jpayne@7
|
84 effort. It can be used in most situations, while still having the
|
jpayne@7
|
85 option to drop down to more specific methods when necessary, such as
|
jpayne@7
|
86 :meth:`request_encode_url`, :meth:`request_encode_body`,
|
jpayne@7
|
87 or even the lowest level :meth:`urlopen`.
|
jpayne@7
|
88
|
jpayne@7
|
89 :param method:
|
jpayne@7
|
90 HTTP request method (such as GET, POST, PUT, etc.)
|
jpayne@7
|
91
|
jpayne@7
|
92 :param url:
|
jpayne@7
|
93 The URL to perform the request on.
|
jpayne@7
|
94
|
jpayne@7
|
95 :param body:
|
jpayne@7
|
96 Data to send in the request body, either :class:`str`, :class:`bytes`,
|
jpayne@7
|
97 an iterable of :class:`str`/:class:`bytes`, or a file-like object.
|
jpayne@7
|
98
|
jpayne@7
|
99 :param fields:
|
jpayne@7
|
100 Data to encode and send in the request body. Values are processed
|
jpayne@7
|
101 by :func:`urllib.parse.urlencode`.
|
jpayne@7
|
102
|
jpayne@7
|
103 :param headers:
|
jpayne@7
|
104 Dictionary of custom headers to send, such as User-Agent,
|
jpayne@7
|
105 If-None-Match, etc. If None, pool headers are used. If provided,
|
jpayne@7
|
106 these headers completely replace any pool-specific headers.
|
jpayne@7
|
107
|
jpayne@7
|
108 :param json:
|
jpayne@7
|
109 Data to encode and send as JSON with UTF-encoded in the request body.
|
jpayne@7
|
110 The ``"Content-Type"`` header will be set to ``"application/json"``
|
jpayne@7
|
111 unless specified otherwise.
|
jpayne@7
|
112 """
|
jpayne@7
|
113 method = method.upper()
|
jpayne@7
|
114
|
jpayne@7
|
115 if json is not None and body is not None:
|
jpayne@7
|
116 raise TypeError(
|
jpayne@7
|
117 "request got values for both 'body' and 'json' parameters which are mutually exclusive"
|
jpayne@7
|
118 )
|
jpayne@7
|
119
|
jpayne@7
|
120 if json is not None:
|
jpayne@7
|
121 if headers is None:
|
jpayne@7
|
122 headers = self.headers
|
jpayne@7
|
123
|
jpayne@7
|
124 if not ("content-type" in map(str.lower, headers.keys())):
|
jpayne@7
|
125 headers = HTTPHeaderDict(headers)
|
jpayne@7
|
126 headers["Content-Type"] = "application/json"
|
jpayne@7
|
127
|
jpayne@7
|
128 body = _json.dumps(json, separators=(",", ":"), ensure_ascii=False).encode(
|
jpayne@7
|
129 "utf-8"
|
jpayne@7
|
130 )
|
jpayne@7
|
131
|
jpayne@7
|
132 if body is not None:
|
jpayne@7
|
133 urlopen_kw["body"] = body
|
jpayne@7
|
134
|
jpayne@7
|
135 if method in self._encode_url_methods:
|
jpayne@7
|
136 return self.request_encode_url(
|
jpayne@7
|
137 method,
|
jpayne@7
|
138 url,
|
jpayne@7
|
139 fields=fields, # type: ignore[arg-type]
|
jpayne@7
|
140 headers=headers,
|
jpayne@7
|
141 **urlopen_kw,
|
jpayne@7
|
142 )
|
jpayne@7
|
143 else:
|
jpayne@7
|
144 return self.request_encode_body(
|
jpayne@7
|
145 method, url, fields=fields, headers=headers, **urlopen_kw
|
jpayne@7
|
146 )
|
jpayne@7
|
147
|
jpayne@7
|
148 def request_encode_url(
|
jpayne@7
|
149 self,
|
jpayne@7
|
150 method: str,
|
jpayne@7
|
151 url: str,
|
jpayne@7
|
152 fields: _TYPE_ENCODE_URL_FIELDS | None = None,
|
jpayne@7
|
153 headers: typing.Mapping[str, str] | None = None,
|
jpayne@7
|
154 **urlopen_kw: str,
|
jpayne@7
|
155 ) -> BaseHTTPResponse:
|
jpayne@7
|
156 """
|
jpayne@7
|
157 Make a request using :meth:`urlopen` with the ``fields`` encoded in
|
jpayne@7
|
158 the url. This is useful for request methods like GET, HEAD, DELETE, etc.
|
jpayne@7
|
159
|
jpayne@7
|
160 :param method:
|
jpayne@7
|
161 HTTP request method (such as GET, POST, PUT, etc.)
|
jpayne@7
|
162
|
jpayne@7
|
163 :param url:
|
jpayne@7
|
164 The URL to perform the request on.
|
jpayne@7
|
165
|
jpayne@7
|
166 :param fields:
|
jpayne@7
|
167 Data to encode and send in the request body.
|
jpayne@7
|
168
|
jpayne@7
|
169 :param headers:
|
jpayne@7
|
170 Dictionary of custom headers to send, such as User-Agent,
|
jpayne@7
|
171 If-None-Match, etc. If None, pool headers are used. If provided,
|
jpayne@7
|
172 these headers completely replace any pool-specific headers.
|
jpayne@7
|
173 """
|
jpayne@7
|
174 if headers is None:
|
jpayne@7
|
175 headers = self.headers
|
jpayne@7
|
176
|
jpayne@7
|
177 extra_kw: dict[str, typing.Any] = {"headers": headers}
|
jpayne@7
|
178 extra_kw.update(urlopen_kw)
|
jpayne@7
|
179
|
jpayne@7
|
180 if fields:
|
jpayne@7
|
181 url += "?" + urlencode(fields)
|
jpayne@7
|
182
|
jpayne@7
|
183 return self.urlopen(method, url, **extra_kw)
|
jpayne@7
|
184
|
jpayne@7
|
185 def request_encode_body(
|
jpayne@7
|
186 self,
|
jpayne@7
|
187 method: str,
|
jpayne@7
|
188 url: str,
|
jpayne@7
|
189 fields: _TYPE_FIELDS | None = None,
|
jpayne@7
|
190 headers: typing.Mapping[str, str] | None = None,
|
jpayne@7
|
191 encode_multipart: bool = True,
|
jpayne@7
|
192 multipart_boundary: str | None = None,
|
jpayne@7
|
193 **urlopen_kw: str,
|
jpayne@7
|
194 ) -> BaseHTTPResponse:
|
jpayne@7
|
195 """
|
jpayne@7
|
196 Make a request using :meth:`urlopen` with the ``fields`` encoded in
|
jpayne@7
|
197 the body. This is useful for request methods like POST, PUT, PATCH, etc.
|
jpayne@7
|
198
|
jpayne@7
|
199 When ``encode_multipart=True`` (default), then
|
jpayne@7
|
200 :func:`urllib3.encode_multipart_formdata` is used to encode
|
jpayne@7
|
201 the payload with the appropriate content type. Otherwise
|
jpayne@7
|
202 :func:`urllib.parse.urlencode` is used with the
|
jpayne@7
|
203 'application/x-www-form-urlencoded' content type.
|
jpayne@7
|
204
|
jpayne@7
|
205 Multipart encoding must be used when posting files, and it's reasonably
|
jpayne@7
|
206 safe to use it in other times too. However, it may break request
|
jpayne@7
|
207 signing, such as with OAuth.
|
jpayne@7
|
208
|
jpayne@7
|
209 Supports an optional ``fields`` parameter of key/value strings AND
|
jpayne@7
|
210 key/filetuple. A filetuple is a (filename, data, MIME type) tuple where
|
jpayne@7
|
211 the MIME type is optional. For example::
|
jpayne@7
|
212
|
jpayne@7
|
213 fields = {
|
jpayne@7
|
214 'foo': 'bar',
|
jpayne@7
|
215 'fakefile': ('foofile.txt', 'contents of foofile'),
|
jpayne@7
|
216 'realfile': ('barfile.txt', open('realfile').read()),
|
jpayne@7
|
217 'typedfile': ('bazfile.bin', open('bazfile').read(),
|
jpayne@7
|
218 'image/jpeg'),
|
jpayne@7
|
219 'nonamefile': 'contents of nonamefile field',
|
jpayne@7
|
220 }
|
jpayne@7
|
221
|
jpayne@7
|
222 When uploading a file, providing a filename (the first parameter of the
|
jpayne@7
|
223 tuple) is optional but recommended to best mimic behavior of browsers.
|
jpayne@7
|
224
|
jpayne@7
|
225 Note that if ``headers`` are supplied, the 'Content-Type' header will
|
jpayne@7
|
226 be overwritten because it depends on the dynamic random boundary string
|
jpayne@7
|
227 which is used to compose the body of the request. The random boundary
|
jpayne@7
|
228 string can be explicitly set with the ``multipart_boundary`` parameter.
|
jpayne@7
|
229
|
jpayne@7
|
230 :param method:
|
jpayne@7
|
231 HTTP request method (such as GET, POST, PUT, etc.)
|
jpayne@7
|
232
|
jpayne@7
|
233 :param url:
|
jpayne@7
|
234 The URL to perform the request on.
|
jpayne@7
|
235
|
jpayne@7
|
236 :param fields:
|
jpayne@7
|
237 Data to encode and send in the request body.
|
jpayne@7
|
238
|
jpayne@7
|
239 :param headers:
|
jpayne@7
|
240 Dictionary of custom headers to send, such as User-Agent,
|
jpayne@7
|
241 If-None-Match, etc. If None, pool headers are used. If provided,
|
jpayne@7
|
242 these headers completely replace any pool-specific headers.
|
jpayne@7
|
243
|
jpayne@7
|
244 :param encode_multipart:
|
jpayne@7
|
245 If True, encode the ``fields`` using the multipart/form-data MIME
|
jpayne@7
|
246 format.
|
jpayne@7
|
247
|
jpayne@7
|
248 :param multipart_boundary:
|
jpayne@7
|
249 If not specified, then a random boundary will be generated using
|
jpayne@7
|
250 :func:`urllib3.filepost.choose_boundary`.
|
jpayne@7
|
251 """
|
jpayne@7
|
252 if headers is None:
|
jpayne@7
|
253 headers = self.headers
|
jpayne@7
|
254
|
jpayne@7
|
255 extra_kw: dict[str, typing.Any] = {"headers": HTTPHeaderDict(headers)}
|
jpayne@7
|
256 body: bytes | str
|
jpayne@7
|
257
|
jpayne@7
|
258 if fields:
|
jpayne@7
|
259 if "body" in urlopen_kw:
|
jpayne@7
|
260 raise TypeError(
|
jpayne@7
|
261 "request got values for both 'fields' and 'body', can only specify one."
|
jpayne@7
|
262 )
|
jpayne@7
|
263
|
jpayne@7
|
264 if encode_multipart:
|
jpayne@7
|
265 body, content_type = encode_multipart_formdata(
|
jpayne@7
|
266 fields, boundary=multipart_boundary
|
jpayne@7
|
267 )
|
jpayne@7
|
268 else:
|
jpayne@7
|
269 body, content_type = (
|
jpayne@7
|
270 urlencode(fields), # type: ignore[arg-type]
|
jpayne@7
|
271 "application/x-www-form-urlencoded",
|
jpayne@7
|
272 )
|
jpayne@7
|
273
|
jpayne@7
|
274 extra_kw["body"] = body
|
jpayne@7
|
275 extra_kw["headers"].setdefault("Content-Type", content_type)
|
jpayne@7
|
276
|
jpayne@7
|
277 extra_kw.update(urlopen_kw)
|
jpayne@7
|
278
|
jpayne@7
|
279 return self.urlopen(method, url, **extra_kw)
|