codekingpro/portable-devtools

1from __future__ import annotations2 3import json as _json4import typing5from urllib.parse import urlencode6 7from ._base_connection import _TYPE_BODY8from ._collections import HTTPHeaderDict9from .filepost import _TYPE_FIELDS, encode_multipart_formdata10from .response import BaseHTTPResponse11 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, such23    as :class:`urllib3.HTTPConnectionPool` and24    :class:`urllib3.PoolManager`.25 26    Provides behavior for making common types of HTTP request methods and27    decides which type of request field encoding to use.28 29    Specifically,30 31    :meth:`.request_encode_url` is for sending requests whose fields are32    encoded in the URL (such as GET, HEAD, DELETE).33 34    :meth:`.request_encode_body` is for sending requests whose fields are35    encoded in the *body* of the request using multipart or www-form-urlencoded36    (such as for POST, PUT, PATCH).37 38    :meth:`.request` is for making any kind of request, it will look up the39    appropriate encoding format and use one of the above two methods to make40    the request.41 42    Initializer parameters:43 44    :param headers:45        Headers to include with all requests, unless other headers are given46        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:  # Abstract64        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 of81        ``fields`` based on the ``method`` used.82 83        This is a convenience method that requires the least amount of manual84        effort. It can be used in most situations, while still having the85        option to drop down to more specific methods when necessary, such as86        :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.headers122 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"] = body133 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_kw145            )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 in157        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.headers175 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 in196        the body. This is useful for request methods like POST, PUT, PATCH, etc.197 198        When ``encode_multipart=True`` (default), then199        :func:`urllib3.encode_multipart_formdata` is used to encode200        the payload with the appropriate content type. Otherwise201        :func:`urllib.parse.urlencode` is used with the202        'application/x-www-form-urlencoded' content type.203 204        Multipart encoding must be used when posting files, and it's reasonably205        safe to use it in other times too. However, it may break request206        signing, such as with OAuth.207 208        Supports an optional ``fields`` parameter of key/value strings AND209        key/filetuple. A filetuple is a (filename, data, MIME type) tuple where210        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 the222        tuple) is optional but recommended to best mimic behavior of browsers.223 224        Note that if ``headers`` are supplied, the 'Content-Type' header will225        be overwritten because it depends on the dynamic random boundary string226        which is used to compose the body of the request. The random boundary227        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 MIME245            format.246 247        :param multipart_boundary:248            If not specified, then a random boundary will be generated using249            :func:`urllib3.filepost.choose_boundary`.250        """251        if headers is None:252            headers = self.headers253 254        extra_kw: dict[str, typing.Any] = {"headers": HTTPHeaderDict(headers)}255        body: bytes | str256 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_boundary266                )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"] = body274            extra_kw["headers"].setdefault("Content-Type", content_type)275 276        extra_kw.update(urlopen_kw)277 278        return self.urlopen(method, url, **extra_kw)279