view env/lib/python3.9/site-packages/requests_toolbelt/ @ 0:4f3585e2f14b draft default tip

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author shellac
date Mon, 22 Mar 2021 18:12:50 +0000
line wrap: on
line source

"""Private module full of compatibility hacks.

Primarily this is for downstream redistributions of requests that unvendor
urllib3 without providing a shim.

.. warning::

    This module is private. If you use it, and something breaks, you were
import sys

import requests

    from requests.packages.urllib3 import fields
    from requests.packages.urllib3 import filepost
    from requests.packages.urllib3 import poolmanager
except ImportError:
    from urllib3 import fields
    from urllib3 import filepost
    from urllib3 import poolmanager

    from requests.packages.urllib3.connection import HTTPConnection
    from requests.packages.urllib3 import connection
except ImportError:
        from urllib3.connection import HTTPConnection
        from urllib3 import connection
    except ImportError:
        HTTPConnection = None
        connection = None

if requests.__build__ < 0x020300:
    timeout = None
        from requests.packages.urllib3.util import timeout
    except ImportError:
        from urllib3.util import timeout

if requests.__build__ < 0x021000:
    gaecontrib = None
        from requests.packages.urllib3.contrib import appengine as gaecontrib
    except ImportError:
        from urllib3.contrib import appengine as gaecontrib

if requests.__build__ < 0x021200:
    PyOpenSSLContext = None
        from requests.packages.urllib3.contrib.pyopenssl \
                import PyOpenSSLContext
    except ImportError:
            from urllib3.contrib.pyopenssl import PyOpenSSLContext
        except ImportError:
            PyOpenSSLContext = None

PY3 = sys.version_info > (3, 0)

if PY3:
    from import Mapping, MutableMapping
    import queue
    from urllib.parse import urlencode, urljoin
    from collections import Mapping, MutableMapping
    import Queue as queue
    from urllib import urlencode
    from urlparse import urljoin

    basestring = basestring
except NameError:
    basestring = (str, bytes)

class HTTPHeaderDict(MutableMapping):
    :param headers:
        An iterable of field-value pairs. Must not contain multiple field names
        when compared case-insensitively.

    :param kwargs:
        Additional field-value pairs to pass in to ``dict.update``.

    A ``dict`` like container for storing HTTP Headers.

    Field names are stored and compared case-insensitively in compliance with
    RFC 7230. Iteration provides the first case-sensitive key seen for each
    case-insensitive pair.

    Using ``__setitem__`` syntax overwrites fields that compare equal
    case-insensitively in order to maintain ``dict``'s api. For fields that
    compare equal, instead create a new ``HTTPHeaderDict`` and use ``.add``
    in a loop.

    If multiple fields that are equal case-insensitively are passed to the
    constructor or ``.update``, the behavior is undefined and some will be

    >>> headers = HTTPHeaderDict()
    >>> headers.add('Set-Cookie', 'foo=bar')
    >>> headers.add('set-cookie', 'baz=quxx')
    >>> headers['content-length'] = '7'
    >>> headers['SET-cookie']
    'foo=bar, baz=quxx'
    >>> headers['Content-Length']

    def __init__(self, headers=None, **kwargs):
        super(HTTPHeaderDict, self).__init__()
        self._container = {}
        if headers is not None:
            if isinstance(headers, HTTPHeaderDict):
        if kwargs:

    def __setitem__(self, key, val):
        self._container[key.lower()] = (key, val)
        return self._container[key.lower()]

    def __getitem__(self, key):
        val = self._container[key.lower()]
        return ', '.join(val[1:])

    def __delitem__(self, key):
        del self._container[key.lower()]

    def __contains__(self, key):
        return key.lower() in self._container

    def __eq__(self, other):
        if not isinstance(other, Mapping) and not hasattr(other, 'keys'):
            return False
        if not isinstance(other, type(self)):
            other = type(self)(other)
        return (dict((k.lower(), v) for k, v in self.itermerged()) ==
                dict((k.lower(), v) for k, v in other.itermerged()))

    def __ne__(self, other):
        return not self.__eq__(other)

    if not PY3:  # Python 2
        iterkeys = MutableMapping.iterkeys
        itervalues = MutableMapping.itervalues

    __marker = object()

    def __len__(self):
        return len(self._container)

    def __iter__(self):
        # Only provide the originally cased names
        for vals in self._container.values():
            yield vals[0]

    def pop(self, key, default=__marker):
        """D.pop(k[,d]) -> v, remove specified key and return its value.

        If key is not found, d is returned if given, otherwise KeyError is
        # Using the MutableMapping function directly fails due to the private
        # marker.
        # Using ordinary dict.pop would expose the internal structures.
        # So let's reinvent the wheel.
            value = self[key]
        except KeyError:
            if default is self.__marker:
            return default
            del self[key]
            return value

    def discard(self, key):
            del self[key]
        except KeyError:

    def add(self, key, val):
        """Adds a (name, value) pair, doesn't overwrite the value if it already

        >>> headers = HTTPHeaderDict(foo='bar')
        >>> headers.add('Foo', 'baz')
        >>> headers['foo']
        'bar, baz'
        key_lower = key.lower()
        new_vals = key, val
        # Keep the common case aka no item present as fast as possible
        vals = self._container.setdefault(key_lower, new_vals)
        if new_vals is not vals:
            # new_vals was not inserted, as there was a previous one
            if isinstance(vals, list):
                # If already several items got inserted, we have a list
                # vals should be a tuple then, i.e. only one item so far
                # Need to convert the tuple to list for further extension
                self._container[key_lower] = [vals[0], vals[1], val]

    def extend(self, *args, **kwargs):
        """Generic import function for any type of header-like object.
        Adapted version of MutableMapping.update in order to insert items
        with self.add instead of self.__setitem__
        if len(args) > 1:
            raise TypeError("extend() takes at most 1 positional "
                            "arguments ({} given)".format(len(args)))
        other = args[0] if len(args) >= 1 else ()

        if isinstance(other, HTTPHeaderDict):
            for key, val in other.iteritems():
                self.add(key, val)
        elif isinstance(other, Mapping):
            for key in other:
                self.add(key, other[key])
        elif hasattr(other, "keys"):
            for key in other.keys():
                self.add(key, other[key])
            for key, value in other:
                self.add(key, value)

        for key, value in kwargs.items():
            self.add(key, value)

    def getlist(self, key):
        """Returns a list of all the values for the named field. Returns an
        empty list if the key doesn't exist."""
            vals = self._container[key.lower()]
        except KeyError:
            return []
            if isinstance(vals, tuple):
                return [vals[1]]
                return vals[1:]

    # Backwards compatibility for httplib
    getheaders = getlist
    getallmatchingheaders = getlist
    iget = getlist

    def __repr__(self):
        return "%s(%s)" % (type(self).__name__, dict(self.itermerged()))

    def _copy_from(self, other):
        for key in other:
            val = other.getlist(key)
            if isinstance(val, list):
                # Don't need to convert tuples
                val = list(val)
            self._container[key.lower()] = [key] + val

    def copy(self):
        clone = type(self)()
        return clone

    def iteritems(self):
        """Iterate over all header lines, including duplicate ones."""
        for key in self:
            vals = self._container[key.lower()]
            for val in vals[1:]:
                yield vals[0], val

    def itermerged(self):
        """Iterate over all headers, merging duplicate ones together."""
        for key in self:
            val = self._container[key.lower()]
            yield val[0], ', '.join(val[1:])

    def items(self):
        return list(self.iteritems())

    def from_httplib(cls, message):  # Python 2
        """Read headers from a Python 2 httplib message object."""
        # python2.7 does not expose a proper API for exporting multiheaders
        # efficiently. This function re-reads raw lines from the message
        # object and extracts the multiheaders properly.
        headers = []

        for line in message.headers:
            if line.startswith((' ', '\t')):
                key, value = headers[-1]
                headers[-1] = (key, value + '\r\n' + line.rstrip())

            key, value = line.split(':', 1)
            headers.append((key, value.strip()))

        return cls(headers)

__all__ = (