venv/lib/python3.11/site-packages/pip/_vendor/urllib3/util/request.py

   1 from __future__ import absolute_import
   2
   3 from base64 import b64encode
   4
   5 from ..exceptions import UnrewindableBodyError
   6 from ..packages.six import b, integer_types
   7
   8 # Pass as a value within ``headers`` to skip
   9 # emitting some HTTP headers that are added automatically.
  10 # The only headers that are supported are ``Accept-Encoding``,
  11 # ``Host``, and ``User-Agent``.
  12 SKIP_HEADER = "@@@SKIP_HEADER@@@"
  13 SKIPPABLE_HEADERS = frozenset(["accept-encoding", "host", "user-agent"])
  14
  15 ACCEPT_ENCODING = "gzip,deflate"
  16
  17 _FAILEDTELL = object()
  18
  19
  20 def make_headers(
  21     keep_alive=None,
  22     accept_encoding=None,
  23     user_agent=None,
  24     basic_auth=None,
  25     proxy_basic_auth=None,
  26     disable_cache=None,
  27 ):
  28     """
  29     Shortcuts for generating request headers.
  30
  31     :param keep_alive:
  32         If ``True``, adds 'connection: keep-alive' header.
  33
  34     :param accept_encoding:
  35         Can be a boolean, list, or string.
  36         ``True`` translates to 'gzip,deflate'.
  37         List will get joined by comma.
  38         String will be used as provided.
  39
  40     :param user_agent:
  41         String representing the user-agent you want, such as
  42         "python-urllib3/0.6"
  43
  44     :param basic_auth:
  45         Colon-separated username:password string for 'authorization: basic ...'
  46         auth header.
  47
  48     :param proxy_basic_auth:
  49         Colon-separated username:password string for 'proxy-authorization: basic ...'
  50         auth header.
  51
  52     :param disable_cache:
  53         If ``True``, adds 'cache-control: no-cache' header.
  54
  55     Example::
  56
  57         >>> make_headers(keep_alive=True, user_agent="Batman/1.0")
  58         {'connection': 'keep-alive', 'user-agent': 'Batman/1.0'}
  59         >>> make_headers(accept_encoding=True)
  60         {'accept-encoding': 'gzip,deflate'}
  61     """
  62     headers = {}
  63     if accept_encoding:
  64         if isinstance(accept_encoding, str):
  65             pass
  66         elif isinstance(accept_encoding, list):
  67             accept_encoding = ",".join(accept_encoding)
  68         else:
  69             accept_encoding = ACCEPT_ENCODING
  70         headers["accept-encoding"] = accept_encoding
  71
  72     if user_agent:
  73         headers["user-agent"] = user_agent
  74
  75     if keep_alive:
  76         headers["connection"] = "keep-alive"
  77
  78     if basic_auth:
  79         headers["authorization"] = "Basic " + b64encode(b(basic_auth)).decode("utf-8")
  80
  81     if proxy_basic_auth:
  82         headers["proxy-authorization"] = "Basic " + b64encode(
  83             b(proxy_basic_auth)
  84         ).decode("utf-8")
  85
  86     if disable_cache:
  87         headers["cache-control"] = "no-cache"
  88
  89     return headers
  90
  91
  92 def set_file_position(body, pos):
  93     """
  94     If a position is provided, move file to that point.
  95     Otherwise, we'll attempt to record a position for future use.
  96     """
  97     if pos is not None:
  98         rewind_body(body, pos)
  99     elif getattr(body, "tell", None) is not None:
 100         try:
 101             pos = body.tell()
 102         except (IOError, OSError):
 103             # This differentiates from None, allowing us to catch
 104             # a failed `tell()` later when trying to rewind the body.
 105             pos = _FAILEDTELL
 106
 107     return pos
 108
 109
 110 def rewind_body(body, body_pos):
 111     """
 112     Attempt to rewind body to a certain position.
 113     Primarily used for request redirects and retries.
 114
 115     :param body:
 116         File-like object that supports seek.
 117
 118     :param int pos:
 119         Position to seek to in file.
 120     """
 121     body_seek = getattr(body, "seek", None)
 122     if body_seek is not None and isinstance(body_pos, integer_types):
 123         try:
 124             body_seek(body_pos)
 125         except (IOError, OSError):
 126             raise UnrewindableBodyError(
 127                 "An error occurred when rewinding request body for redirect/retry."
 128             )
 129     elif body_pos is _FAILEDTELL:
 130         raise UnrewindableBodyError(
 131             "Unable to record file position for rewinding "
 132             "request body during a redirect/retry."
 133         )
 134     else:
 135         raise ValueError(
 136             "body_pos must be of type integer, instead it was %s." % type(body_pos)
 137         )