FM / util

/usr/local/lib/python3.6/site-packages/pip/_vendor/urllib3/util/

/usr/local/lib/python3.6/site-packages/pip/_vendor/urllib3/util

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
connection.py	4920	0644	edit dl rm
proxy.py	1605	0644	edit dl rm
queue.py	498	0644	edit dl rm
request.py	4123	0644	edit dl rm
response.py	3510	0644	edit dl rm
retry.py	21391	0644	edit dl rm
ssltransport.py	6931	0644	edit dl rm
ssl_.py	17177	0644	edit dl rm
timeout.py	10003	0644	edit dl rm
url.py	14047	0644	edit dl rm
wait.py	5404	0644	edit dl rm
__init__.py	1155	0644	edit dl rm

Edit: /usr/local/lib/python3.6/site-packages/pip/_vendor/urllib3/util/request.py (4123B)

from __future__ import absolute_import

from base64 import b64encode

from ..exceptions import UnrewindableBodyError
from ..packages.six import b, integer_types

# Pass as a value within ``headers`` to skip
# emitting some HTTP headers that are added automatically.
# The only headers that are supported are ``Accept-Encoding``,
# ``Host``, and ``User-Agent``.
SKIP_HEADER = "@@@SKIP_HEADER@@@"
SKIPPABLE_HEADERS = frozenset(["accept-encoding", "host", "user-agent"])

ACCEPT_ENCODING = "gzip,deflate"
try:
    import brotli as _unused_module_brotli  # noqa: F401
except ImportError:
    pass
else:
    ACCEPT_ENCODING += ",br"

_FAILEDTELL = object()

def make_headers(
    keep_alive=None,
    accept_encoding=None,
    user_agent=None,
    basic_auth=None,
    proxy_basic_auth=None,
    disable_cache=None,
):
    """
    Shortcuts for generating request headers.

:param keep_alive:
        If ``True``, adds 'connection: keep-alive' header.

:param accept_encoding:
        Can be a boolean, list, or string.
        ``True`` translates to 'gzip,deflate'.
        List will get joined by comma.
        String will be used as provided.

:param user_agent:
        String representing the user-agent you want, such as
        "python-urllib3/0.6"

:param basic_auth:
        Colon-separated username:password string for 'authorization: basic ...'
        auth header.

:param proxy_basic_auth:
        Colon-separated username:password string for 'proxy-authorization: basic ...'
        auth header.

:param disable_cache:
        If ``True``, adds 'cache-control: no-cache' header.

Example::

>>> make_headers(keep_alive=True, user_agent="Batman/1.0")
        {'connection': 'keep-alive', 'user-agent': 'Batman/1.0'}
        >>> make_headers(accept_encoding=True)
        {'accept-encoding': 'gzip,deflate'}
    """
    headers = {}
    if accept_encoding:
        if isinstance(accept_encoding, str):
            pass
        elif isinstance(accept_encoding, list):
            accept_encoding = ",".join(accept_encoding)
        else:
            accept_encoding = ACCEPT_ENCODING
        headers["accept-encoding"] = accept_encoding

if user_agent:
        headers["user-agent"] = user_agent

if keep_alive:
        headers["connection"] = "keep-alive"

if basic_auth:
        headers["authorization"] = "Basic " + b64encode(b(basic_auth)).decode("utf-8")

if proxy_basic_auth:
        headers["proxy-authorization"] = "Basic " + b64encode(
            b(proxy_basic_auth)
        ).decode("utf-8")

if disable_cache:
        headers["cache-control"] = "no-cache"

return headers

def set_file_position(body, pos):
    """
    If a position is provided, move file to that point.
    Otherwise, we'll attempt to record a position for future use.
    """
    if pos is not None:
        rewind_body(body, pos)
    elif getattr(body, "tell", None) is not None:
        try:
            pos = body.tell()
        except (IOError, OSError):
            # This differentiates from None, allowing us to catch
            # a failed `tell()` later when trying to rewind the body.
            pos = _FAILEDTELL

return pos

def rewind_body(body, body_pos):
    """
    Attempt to rewind body to a certain position.
    Primarily used for request redirects and retries.

:param body:
        File-like object that supports seek.

:param int pos:
        Position to seek to in file.
    """
    body_seek = getattr(body, "seek", None)
    if body_seek is not None and isinstance(body_pos, integer_types):
        try:
            body_seek(body_pos)
        except (IOError, OSError):
            raise UnrewindableBodyError(
                "An error occurred when rewinding request body for redirect/retry."
            )
    elif body_pos is _FAILEDTELL:
        raise UnrewindableBodyError(
            "Unable to record file position for rewinding "
            "request body during a redirect/retry."
        )
    else:
        raise ValueError(
            "body_pos must be of type integer, instead it was %s." % type(body_pos)
        )

cmd: