Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/urllib3/util/request.py: 48%
55 statements
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
1from __future__ import absolute_import
3from base64 import b64encode
5from ..exceptions import UnrewindableBodyError
6from ..packages.six import b, integer_types
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``.
12SKIP_HEADER = "@@@SKIP_HEADER@@@"
13SKIPPABLE_HEADERS = frozenset(["accept-encoding", "host", "user-agent"])
15ACCEPT_ENCODING = "gzip,deflate"
16try:
17 try:
18 import brotlicffi as _unused_module_brotli # noqa: F401
19 except ImportError:
20 import brotli as _unused_module_brotli # noqa: F401
21except ImportError:
22 pass
23else:
24 ACCEPT_ENCODING += ",br"
26_FAILEDTELL = object()
29def make_headers(
30 keep_alive=None,
31 accept_encoding=None,
32 user_agent=None,
33 basic_auth=None,
34 proxy_basic_auth=None,
35 disable_cache=None,
36):
37 """
38 Shortcuts for generating request headers.
40 :param keep_alive:
41 If ``True``, adds 'connection: keep-alive' header.
43 :param accept_encoding:
44 Can be a boolean, list, or string.
45 ``True`` translates to 'gzip,deflate'.
46 List will get joined by comma.
47 String will be used as provided.
49 :param user_agent:
50 String representing the user-agent you want, such as
51 "python-urllib3/0.6"
53 :param basic_auth:
54 Colon-separated username:password string for 'authorization: basic ...'
55 auth header.
57 :param proxy_basic_auth:
58 Colon-separated username:password string for 'proxy-authorization: basic ...'
59 auth header.
61 :param disable_cache:
62 If ``True``, adds 'cache-control: no-cache' header.
64 Example::
66 >>> make_headers(keep_alive=True, user_agent="Batman/1.0")
67 {'connection': 'keep-alive', 'user-agent': 'Batman/1.0'}
68 >>> make_headers(accept_encoding=True)
69 {'accept-encoding': 'gzip,deflate'}
70 """
71 headers = {}
72 if accept_encoding: 72 ↛ 81line 72 didn't jump to line 81, because the condition on line 72 was never false
73 if isinstance(accept_encoding, str): 73 ↛ 74line 73 didn't jump to line 74, because the condition on line 73 was never true
74 pass
75 elif isinstance(accept_encoding, list): 75 ↛ 76line 75 didn't jump to line 76, because the condition on line 75 was never true
76 accept_encoding = ",".join(accept_encoding)
77 else:
78 accept_encoding = ACCEPT_ENCODING
79 headers["accept-encoding"] = accept_encoding
81 if user_agent: 81 ↛ 82line 81 didn't jump to line 82, because the condition on line 81 was never true
82 headers["user-agent"] = user_agent
84 if keep_alive: 84 ↛ 85line 84 didn't jump to line 85, because the condition on line 84 was never true
85 headers["connection"] = "keep-alive"
87 if basic_auth: 87 ↛ 88line 87 didn't jump to line 88, because the condition on line 87 was never true
88 headers["authorization"] = "Basic " + b64encode(b(basic_auth)).decode("utf-8")
90 if proxy_basic_auth: 90 ↛ 91line 90 didn't jump to line 91, because the condition on line 90 was never true
91 headers["proxy-authorization"] = "Basic " + b64encode(
92 b(proxy_basic_auth)
93 ).decode("utf-8")
95 if disable_cache: 95 ↛ 96line 95 didn't jump to line 96, because the condition on line 95 was never true
96 headers["cache-control"] = "no-cache"
98 return headers
101def set_file_position(body, pos):
102 """
103 If a position is provided, move file to that point.
104 Otherwise, we'll attempt to record a position for future use.
105 """
106 if pos is not None:
107 rewind_body(body, pos)
108 elif getattr(body, "tell", None) is not None:
109 try:
110 pos = body.tell()
111 except (IOError, OSError):
112 # This differentiates from None, allowing us to catch
113 # a failed `tell()` later when trying to rewind the body.
114 pos = _FAILEDTELL
116 return pos
119def rewind_body(body, body_pos):
120 """
121 Attempt to rewind body to a certain position.
122 Primarily used for request redirects and retries.
124 :param body:
125 File-like object that supports seek.
127 :param int pos:
128 Position to seek to in file.
129 """
130 body_seek = getattr(body, "seek", None)
131 if body_seek is not None and isinstance(body_pos, integer_types):
132 try:
133 body_seek(body_pos)
134 except (IOError, OSError):
135 raise UnrewindableBodyError(
136 "An error occurred when rewinding request body for redirect/retry."
137 )
138 elif body_pos is _FAILEDTELL:
139 raise UnrewindableBodyError(
140 "Unable to record file position for rewinding "
141 "request body during a redirect/retry."
142 )
143 else:
144 raise ValueError(
145 "body_pos must be of type integer, instead it was %s." % type(body_pos)
146 )