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)