144 lines
4.0 KiB
Python
144 lines
4.0 KiB
Python
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)
|
|
)
|