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