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[typing.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 request body. Values are processed
101 by :func:`urllib.parse.urlencode`.
102
103 :param headers:
104 Dictionary of custom headers to send, such as User-Agent,
105 If-None-Match, etc. If None, pool headers are used. If provided,
106 these headers completely replace any pool-specific headers.
107
108 :param json:
109 Data to encode and send as JSON with UTF-encoded in the request body.
110 The ``"Content-Type"`` header will be set to ``"application/json"``
111 unless specified otherwise.
112 """
113 method = method.upper()
114
115 if json is not None and body is not None:
116 raise TypeError(
117 "request got values for both 'body' and 'json' parameters which are mutually exclusive"
118 )
119
120 if json is not None:
121 if headers is None:
122 headers = self.headers
123
124 if not ("content-type" in map(str.lower, headers.keys())):
125 headers = HTTPHeaderDict(headers)
126 headers["Content-Type"] = "application/json"
127
128 body = _json.dumps(json, separators=(",", ":"), ensure_ascii=False).encode(
129 "utf-8"
130 )
131
132 if body is not None:
133 urlopen_kw["body"] = body
134
135 if method in self._encode_url_methods:
136 return self.request_encode_url(
137 method,
138 url,
139 fields=fields, # type: ignore[arg-type]
140 headers=headers,
141 **urlopen_kw,
142 )
143 else:
144 return self.request_encode_body(
145 method, url, fields=fields, headers=headers, **urlopen_kw
146 )
147
148 def request_encode_url(
149 self,
150 method: str,
151 url: str,
152 fields: _TYPE_ENCODE_URL_FIELDS | None = None,
153 headers: typing.Mapping[str, str] | None = None,
154 **urlopen_kw: str,
155 ) -> BaseHTTPResponse:
156 """
157 Make a request using :meth:`urlopen` with the ``fields`` encoded in
158 the url. This is useful for request methods like GET, HEAD, DELETE, etc.
159
160 :param method:
161 HTTP request method (such as GET, POST, PUT, etc.)
162
163 :param url:
164 The URL to perform the request on.
165
166 :param fields:
167 Data to encode and send in the request body.
168
169 :param headers:
170 Dictionary of custom headers to send, such as User-Agent,
171 If-None-Match, etc. If None, pool headers are used. If provided,
172 these headers completely replace any pool-specific headers.
173 """
174 if headers is None:
175 headers = self.headers
176
177 extra_kw: dict[str, typing.Any] = {"headers": headers}
178 extra_kw.update(urlopen_kw)
179
180 if fields:
181 url += "?" + urlencode(fields)
182
183 return self.urlopen(method, url, **extra_kw)
184
185 def request_encode_body(
186 self,
187 method: str,
188 url: str,
189 fields: _TYPE_FIELDS | None = None,
190 headers: typing.Mapping[str, str] | None = None,
191 encode_multipart: bool = True,
192 multipart_boundary: str | None = None,
193 **urlopen_kw: str,
194 ) -> BaseHTTPResponse:
195 """
196 Make a request using :meth:`urlopen` with the ``fields`` encoded in
197 the body. This is useful for request methods like POST, PUT, PATCH, etc.
198
199 When ``encode_multipart=True`` (default), then
200 :func:`urllib3.encode_multipart_formdata` is used to encode
201 the payload with the appropriate content type. Otherwise
202 :func:`urllib.parse.urlencode` is used with the
203 'application/x-www-form-urlencoded' content type.
204
205 Multipart encoding must be used when posting files, and it's reasonably
206 safe to use it in other times too. However, it may break request
207 signing, such as with OAuth.
208
209 Supports an optional ``fields`` parameter of key/value strings AND
210 key/filetuple. A filetuple is a (filename, data, MIME type) tuple where
211 the MIME type is optional. For example::
212
213 fields = {
214 'foo': 'bar',
215 'fakefile': ('foofile.txt', 'contents of foofile'),
216 'realfile': ('barfile.txt', open('realfile').read()),
217 'typedfile': ('bazfile.bin', open('bazfile').read(),
218 'image/jpeg'),
219 'nonamefile': 'contents of nonamefile field',
220 }
221
222 When uploading a file, providing a filename (the first parameter of the
223 tuple) is optional but recommended to best mimic behavior of browsers.
224
225 Note that if ``headers`` are supplied, the 'Content-Type' header will
226 be overwritten because it depends on the dynamic random boundary string
227 which is used to compose the body of the request. The random boundary
228 string can be explicitly set with the ``multipart_boundary`` parameter.
229
230 :param method:
231 HTTP request method (such as GET, POST, PUT, etc.)
232
233 :param url:
234 The URL to perform the request on.
235
236 :param fields:
237 Data to encode and send in the request body.
238
239 :param headers:
240 Dictionary of custom headers to send, such as User-Agent,
241 If-None-Match, etc. If None, pool headers are used. If provided,
242 these headers completely replace any pool-specific headers.
243
244 :param encode_multipart:
245 If True, encode the ``fields`` using the multipart/form-data MIME
246 format.
247
248 :param multipart_boundary:
249 If not specified, then a random boundary will be generated using
250 :func:`urllib3.filepost.choose_boundary`.
251 """
252 if headers is None:
253 headers = self.headers
254
255 extra_kw: dict[str, typing.Any] = {"headers": HTTPHeaderDict(headers)}
256 body: bytes | str
257
258 if fields:
259 if "body" in urlopen_kw:
260 raise TypeError(
261 "request got values for both 'fields' and 'body', can only specify one."
262 )
263
264 if encode_multipart:
265 body, content_type = encode_multipart_formdata(
266 fields, boundary=multipart_boundary
267 )
268 else:
269 body, content_type = (
270 urlencode(fields), # type: ignore[arg-type]
271 "application/x-www-form-urlencoded",
272 )
273
274 extra_kw["body"] = body
275 extra_kw["headers"].setdefault("Content-Type", content_type)
276
277 extra_kw.update(urlopen_kw)
278
279 return self.urlopen(method, url, **extra_kw)