jpayne@7: from __future__ import annotations
jpayne@7: 
jpayne@7: import json as _json
jpayne@7: import typing
jpayne@7: from urllib.parse import urlencode
jpayne@7: 
jpayne@7: from ._base_connection import _TYPE_BODY
jpayne@7: from ._collections import HTTPHeaderDict
jpayne@7: from .filepost import _TYPE_FIELDS, encode_multipart_formdata
jpayne@7: from .response import BaseHTTPResponse
jpayne@7: 
jpayne@7: __all__ = ["RequestMethods"]
jpayne@7: 
jpayne@7: _TYPE_ENCODE_URL_FIELDS = typing.Union[
jpayne@7:     typing.Sequence[typing.Tuple[str, typing.Union[str, bytes]]],
jpayne@7:     typing.Mapping[str, typing.Union[str, bytes]],
jpayne@7: ]
jpayne@7: 
jpayne@7: 
jpayne@7: class RequestMethods:
jpayne@7:     """
jpayne@7:     Convenience mixin for classes who implement a :meth:`urlopen` method, such
jpayne@7:     as :class:`urllib3.HTTPConnectionPool` and
jpayne@7:     :class:`urllib3.PoolManager`.
jpayne@7: 
jpayne@7:     Provides behavior for making common types of HTTP request methods and
jpayne@7:     decides which type of request field encoding to use.
jpayne@7: 
jpayne@7:     Specifically,
jpayne@7: 
jpayne@7:     :meth:`.request_encode_url` is for sending requests whose fields are
jpayne@7:     encoded in the URL (such as GET, HEAD, DELETE).
jpayne@7: 
jpayne@7:     :meth:`.request_encode_body` is for sending requests whose fields are
jpayne@7:     encoded in the *body* of the request using multipart or www-form-urlencoded
jpayne@7:     (such as for POST, PUT, PATCH).
jpayne@7: 
jpayne@7:     :meth:`.request` is for making any kind of request, it will look up the
jpayne@7:     appropriate encoding format and use one of the above two methods to make
jpayne@7:     the request.
jpayne@7: 
jpayne@7:     Initializer parameters:
jpayne@7: 
jpayne@7:     :param headers:
jpayne@7:         Headers to include with all requests, unless other headers are given
jpayne@7:         explicitly.
jpayne@7:     """
jpayne@7: 
jpayne@7:     _encode_url_methods = {"DELETE", "GET", "HEAD", "OPTIONS"}
jpayne@7: 
jpayne@7:     def __init__(self, headers: typing.Mapping[str, str] | None = None) -> None:
jpayne@7:         self.headers = headers or {}
jpayne@7: 
jpayne@7:     def urlopen(
jpayne@7:         self,
jpayne@7:         method: str,
jpayne@7:         url: str,
jpayne@7:         body: _TYPE_BODY | None = None,
jpayne@7:         headers: typing.Mapping[str, str] | None = None,
jpayne@7:         encode_multipart: bool = True,
jpayne@7:         multipart_boundary: str | None = None,
jpayne@7:         **kw: typing.Any,
jpayne@7:     ) -> BaseHTTPResponse:  # Abstract
jpayne@7:         raise NotImplementedError(
jpayne@7:             "Classes extending RequestMethods must implement "
jpayne@7:             "their own ``urlopen`` method."
jpayne@7:         )
jpayne@7: 
jpayne@7:     def request(
jpayne@7:         self,
jpayne@7:         method: str,
jpayne@7:         url: str,
jpayne@7:         body: _TYPE_BODY | None = None,
jpayne@7:         fields: _TYPE_FIELDS | None = None,
jpayne@7:         headers: typing.Mapping[str, str] | None = None,
jpayne@7:         json: typing.Any | None = None,
jpayne@7:         **urlopen_kw: typing.Any,
jpayne@7:     ) -> BaseHTTPResponse:
jpayne@7:         """
jpayne@7:         Make a request using :meth:`urlopen` with the appropriate encoding of
jpayne@7:         ``fields`` based on the ``method`` used.
jpayne@7: 
jpayne@7:         This is a convenience method that requires the least amount of manual
jpayne@7:         effort. It can be used in most situations, while still having the
jpayne@7:         option to drop down to more specific methods when necessary, such as
jpayne@7:         :meth:`request_encode_url`, :meth:`request_encode_body`,
jpayne@7:         or even the lowest level :meth:`urlopen`.
jpayne@7: 
jpayne@7:         :param method:
jpayne@7:             HTTP request method (such as GET, POST, PUT, etc.)
jpayne@7: 
jpayne@7:         :param url:
jpayne@7:             The URL to perform the request on.
jpayne@7: 
jpayne@7:         :param body:
jpayne@7:             Data to send in the request body, either :class:`str`, :class:`bytes`,
jpayne@7:             an iterable of :class:`str`/:class:`bytes`, or a file-like object.
jpayne@7: 
jpayne@7:         :param fields:
jpayne@7:             Data to encode and send in the request body.  Values are processed
jpayne@7:             by :func:`urllib.parse.urlencode`.
jpayne@7: 
jpayne@7:         :param headers:
jpayne@7:             Dictionary of custom headers to send, such as User-Agent,
jpayne@7:             If-None-Match, etc. If None, pool headers are used. If provided,
jpayne@7:             these headers completely replace any pool-specific headers.
jpayne@7: 
jpayne@7:         :param json:
jpayne@7:             Data to encode and send as JSON with UTF-encoded in the request body.
jpayne@7:             The ``"Content-Type"`` header will be set to ``"application/json"``
jpayne@7:             unless specified otherwise.
jpayne@7:         """
jpayne@7:         method = method.upper()
jpayne@7: 
jpayne@7:         if json is not None and body is not None:
jpayne@7:             raise TypeError(
jpayne@7:                 "request got values for both 'body' and 'json' parameters which are mutually exclusive"
jpayne@7:             )
jpayne@7: 
jpayne@7:         if json is not None:
jpayne@7:             if headers is None:
jpayne@7:                 headers = self.headers
jpayne@7: 
jpayne@7:             if not ("content-type" in map(str.lower, headers.keys())):
jpayne@7:                 headers = HTTPHeaderDict(headers)
jpayne@7:                 headers["Content-Type"] = "application/json"
jpayne@7: 
jpayne@7:             body = _json.dumps(json, separators=(",", ":"), ensure_ascii=False).encode(
jpayne@7:                 "utf-8"
jpayne@7:             )
jpayne@7: 
jpayne@7:         if body is not None:
jpayne@7:             urlopen_kw["body"] = body
jpayne@7: 
jpayne@7:         if method in self._encode_url_methods:
jpayne@7:             return self.request_encode_url(
jpayne@7:                 method,
jpayne@7:                 url,
jpayne@7:                 fields=fields,  # type: ignore[arg-type]
jpayne@7:                 headers=headers,
jpayne@7:                 **urlopen_kw,
jpayne@7:             )
jpayne@7:         else:
jpayne@7:             return self.request_encode_body(
jpayne@7:                 method, url, fields=fields, headers=headers, **urlopen_kw
jpayne@7:             )
jpayne@7: 
jpayne@7:     def request_encode_url(
jpayne@7:         self,
jpayne@7:         method: str,
jpayne@7:         url: str,
jpayne@7:         fields: _TYPE_ENCODE_URL_FIELDS | None = None,
jpayne@7:         headers: typing.Mapping[str, str] | None = None,
jpayne@7:         **urlopen_kw: str,
jpayne@7:     ) -> BaseHTTPResponse:
jpayne@7:         """
jpayne@7:         Make a request using :meth:`urlopen` with the ``fields`` encoded in
jpayne@7:         the url. This is useful for request methods like GET, HEAD, DELETE, etc.
jpayne@7: 
jpayne@7:         :param method:
jpayne@7:             HTTP request method (such as GET, POST, PUT, etc.)
jpayne@7: 
jpayne@7:         :param url:
jpayne@7:             The URL to perform the request on.
jpayne@7: 
jpayne@7:         :param fields:
jpayne@7:             Data to encode and send in the request body.
jpayne@7: 
jpayne@7:         :param headers:
jpayne@7:             Dictionary of custom headers to send, such as User-Agent,
jpayne@7:             If-None-Match, etc. If None, pool headers are used. If provided,
jpayne@7:             these headers completely replace any pool-specific headers.
jpayne@7:         """
jpayne@7:         if headers is None:
jpayne@7:             headers = self.headers
jpayne@7: 
jpayne@7:         extra_kw: dict[str, typing.Any] = {"headers": headers}
jpayne@7:         extra_kw.update(urlopen_kw)
jpayne@7: 
jpayne@7:         if fields:
jpayne@7:             url += "?" + urlencode(fields)
jpayne@7: 
jpayne@7:         return self.urlopen(method, url, **extra_kw)
jpayne@7: 
jpayne@7:     def request_encode_body(
jpayne@7:         self,
jpayne@7:         method: str,
jpayne@7:         url: str,
jpayne@7:         fields: _TYPE_FIELDS | None = None,
jpayne@7:         headers: typing.Mapping[str, str] | None = None,
jpayne@7:         encode_multipart: bool = True,
jpayne@7:         multipart_boundary: str | None = None,
jpayne@7:         **urlopen_kw: str,
jpayne@7:     ) -> BaseHTTPResponse:
jpayne@7:         """
jpayne@7:         Make a request using :meth:`urlopen` with the ``fields`` encoded in
jpayne@7:         the body. This is useful for request methods like POST, PUT, PATCH, etc.
jpayne@7: 
jpayne@7:         When ``encode_multipart=True`` (default), then
jpayne@7:         :func:`urllib3.encode_multipart_formdata` is used to encode
jpayne@7:         the payload with the appropriate content type. Otherwise
jpayne@7:         :func:`urllib.parse.urlencode` is used with the
jpayne@7:         'application/x-www-form-urlencoded' content type.
jpayne@7: 
jpayne@7:         Multipart encoding must be used when posting files, and it's reasonably
jpayne@7:         safe to use it in other times too. However, it may break request
jpayne@7:         signing, such as with OAuth.
jpayne@7: 
jpayne@7:         Supports an optional ``fields`` parameter of key/value strings AND
jpayne@7:         key/filetuple. A filetuple is a (filename, data, MIME type) tuple where
jpayne@7:         the MIME type is optional. For example::
jpayne@7: 
jpayne@7:             fields = {
jpayne@7:                 'foo': 'bar',
jpayne@7:                 'fakefile': ('foofile.txt', 'contents of foofile'),
jpayne@7:                 'realfile': ('barfile.txt', open('realfile').read()),
jpayne@7:                 'typedfile': ('bazfile.bin', open('bazfile').read(),
jpayne@7:                               'image/jpeg'),
jpayne@7:                 'nonamefile': 'contents of nonamefile field',
jpayne@7:             }
jpayne@7: 
jpayne@7:         When uploading a file, providing a filename (the first parameter of the
jpayne@7:         tuple) is optional but recommended to best mimic behavior of browsers.
jpayne@7: 
jpayne@7:         Note that if ``headers`` are supplied, the 'Content-Type' header will
jpayne@7:         be overwritten because it depends on the dynamic random boundary string
jpayne@7:         which is used to compose the body of the request. The random boundary
jpayne@7:         string can be explicitly set with the ``multipart_boundary`` parameter.
jpayne@7: 
jpayne@7:         :param method:
jpayne@7:             HTTP request method (such as GET, POST, PUT, etc.)
jpayne@7: 
jpayne@7:         :param url:
jpayne@7:             The URL to perform the request on.
jpayne@7: 
jpayne@7:         :param fields:
jpayne@7:             Data to encode and send in the request body.
jpayne@7: 
jpayne@7:         :param headers:
jpayne@7:             Dictionary of custom headers to send, such as User-Agent,
jpayne@7:             If-None-Match, etc. If None, pool headers are used. If provided,
jpayne@7:             these headers completely replace any pool-specific headers.
jpayne@7: 
jpayne@7:         :param encode_multipart:
jpayne@7:             If True, encode the ``fields`` using the multipart/form-data MIME
jpayne@7:             format.
jpayne@7: 
jpayne@7:         :param multipart_boundary:
jpayne@7:             If not specified, then a random boundary will be generated using
jpayne@7:             :func:`urllib3.filepost.choose_boundary`.
jpayne@7:         """
jpayne@7:         if headers is None:
jpayne@7:             headers = self.headers
jpayne@7: 
jpayne@7:         extra_kw: dict[str, typing.Any] = {"headers": HTTPHeaderDict(headers)}
jpayne@7:         body: bytes | str
jpayne@7: 
jpayne@7:         if fields:
jpayne@7:             if "body" in urlopen_kw:
jpayne@7:                 raise TypeError(
jpayne@7:                     "request got values for both 'fields' and 'body', can only specify one."
jpayne@7:                 )
jpayne@7: 
jpayne@7:             if encode_multipart:
jpayne@7:                 body, content_type = encode_multipart_formdata(
jpayne@7:                     fields, boundary=multipart_boundary
jpayne@7:                 )
jpayne@7:             else:
jpayne@7:                 body, content_type = (
jpayne@7:                     urlencode(fields),  # type: ignore[arg-type]
jpayne@7:                     "application/x-www-form-urlencoded",
jpayne@7:                 )
jpayne@7: 
jpayne@7:             extra_kw["body"] = body
jpayne@7:             extra_kw["headers"].setdefault("Content-Type", content_type)
jpayne@7: 
jpayne@7:         extra_kw.update(urlopen_kw)
jpayne@7: 
jpayne@7:         return self.urlopen(method, url, **extra_kw)