Spaces:
Sleeping
Sleeping
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" | |
_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) | |
) | |