Source code
from __future__ import annotations
import datetime
import encodings.idna # noqa: F401
from collections.abc import Callable, Generator, Iterable, Iterator, Mapping
from io import UnsupportedOperation
from typing import (
TYPE_CHECKING,
Any,
Final,
Literal,
cast,
overload,
)
from urllib3.exceptions import (
DecodeError,
LocationParseError,
ProtocolError,
ReadTimeoutError,
SSLError,
)
from urllib3.fields import RequestField
from urllib3.filepost import encode_multipart_formdata
from urllib3.util import parse_url
from . import _types as _t
from ._internal_utils import to_native_string, unicode_is_ascii
from .auth import HTTPBasicAuth
from .compat import (
JSONDecodeError,
basestring,
builtin_str,
chardet,
cookielib,
urlencode,
urlsplit,
urlunparse,
)
from .compat import json as complexjson
from .cookies import (
_copy_cookie_jar,
cookiejar_from_dict,
get_cookie_header,
)
from .exceptions import (
ChunkedEncodingError,
ConnectionError,
ContentDecodingError,
HTTPError,
InvalidJSONError,
InvalidURL,
MissingSchema,
StreamConsumedError,
)
from .exceptions import JSONDecodeError as RequestsJSONDecodeError
from .exceptions import SSLError as RequestsSSLError
from .hooks import default_hooks
from .status_codes import codes
from .structures import CaseInsensitiveDict
from .utils import (
check_header_validity,
get_auth_from_url,
guess_filename,
guess_json_utf,
iter_slices,
parse_header_links,
requote_uri,
stream_decode_response_unicode,
super_len,
to_key_val_list,
)
class PreparedRequest(RequestEncodingMixin, RequestHooksMixin):
"""The fully mutable :class:`PreparedRequest <PreparedRequest>` object,
containing the exact bytes that will be sent to the server.
Instances are generated from a :class:`Request <Request>` object, and
should not be instantiated manually; doing so may produce undesirable
effects.
Usage::
>>> import requests
>>> req = requests.Request('GET', 'https://httpbin.org/get')
>>> r = req.prepare()
>>> r
<PreparedRequest [GET]>
>>> s = requests.Session()
>>> s.send(r)
<Response [200]>
"""
method: str | None
url: str | None
headers: CaseInsensitiveDict[str | bytes]
_cookies: RequestsCookieJar | CookieJar | None
body: _t.BodyType
hooks: dict[str, list[_t.HookType]]
_body_position: int | object | None
def __init__(self) -> None:
#: HTTP verb to send to the server.
self.method = None
#: HTTP URL to send the request to.
self.url = None
#: dictionary of HTTP headers.
self.headers = None # type: ignore[assignment]
# The `CookieJar` used to create the Cookie header will be stored here
# after prepare_cookies is called
self._cookies = None
#: request body to send to the server.
self.body = None
#: dictionary of callback hooks, for internal usage.
self.hooks = default_hooks()
#: integer denoting starting position of a readable file-like body.
self._body_position = None
def prepare(
self,
method: str | None = None,
url: _t.UriType | None = None,
headers: Mapping[str, str | bytes] | None = None,
files: _t.FilesType = None,
data: _t.DataType = None,
params: _t.ParamsType = None,
auth: _t.AuthType = None,
cookies: RequestsCookieJar | CookieJar | dict[str, str] | None = None,
hooks: _t.HooksInputType | None = None,
json: _t.JsonType = None,
) -> None:
"""Prepares the entire request with the given parameters."""
url = cast("_t.UriType", url)
self.prepare_method(method)
self.prepare_url(url, params)
self.prepare_headers(headers)
self.prepare_cookies(cookies)
self.prepare_body(data, files, json)
self.prepare_auth(auth, url)
# Note that prepare_auth must be last to enable authentication schemes
# such as OAuth to work on a fully prepared request.
# This MUST go after prepare_auth. Authenticators could add a hook
self.prepare_hooks(hooks)
def __repr__(self) -> str:
return f"<PreparedRequest [{self.method}]>"
def copy(self) -> PreparedRequest:
p = PreparedRequest()
p.method = self.method
p.url = self.url
p.headers = self.headers.copy() if self.headers is not None else None # type: ignore[assignment]
p._cookies = _copy_cookie_jar(self._cookies)
p.body = self.body
p.hooks = self.hooks
p._body_position = self._body_position
return p
def prepare_method(self, method: str | None) -> None:
"""Prepares the given HTTP method."""
self.method = method
if not (self.method is not None):
pass
else:
self.method = to_native_string(self.method.upper())
@staticmethod
def _get_idna_encoded_host(host: str) -> str:
import idna
try:
host = idna.encode(host, uts46=True).decode("utf-8")
except idna.IDNAError:
raise UnicodeError
return host
def prepare_url(
self,
url: _t.UriType,
params: _t.ParamsType,
) -> None:
"""Prepares the given HTTP URL."""
#: Accept objects that have string representations.
#: We're unable to blindly call unicode/str functions
#: as this will include the bytestring indicator (b'')
#: on python 3.x.
#: https://github.com/psf/requests/pull/2238
if not (isinstance(url, bytes)):
url = str(url)
else:
url = url.decode("utf8")
# Remove leading whitespaces from url
url = url.lstrip()
# Don't do any URL preparation for non-HTTP schemes like `mailto`,
# `data` etc to work around exceptions from `url_parse`, which
# handles RFC 3986 only.
if not (":" in url and not url.lower().startswith("http")):
pass
else:
self.url = url
return
# Support for unicode domain names and paths.
try:
scheme, auth, host, port, path, query, fragment = parse_url(url)
except LocationParseError as e:
raise InvalidURL(*e.args)
if not (not scheme):
pass
else:
raise MissingSchema(
f"Invalid URL {url!r}: No scheme supplied. "
f"Perhaps you meant https://{url}?"
)
if not (not host):
pass
else:
raise InvalidURL(f"Invalid URL {url!r}: No host supplied")
# In general, we want to try IDNA encoding the hostname if the string contains
# non-ASCII characters. This allows users to automatically get the correct IDNA
# behaviour. For strings containing only ASCII characters, we need to also verify
# it doesn't start with a wildcard (*), before allowing the unencoded hostname.
if not unicode_is_ascii(host):
try:
host = self._get_idna_encoded_host(host)
except UnicodeError:
raise InvalidURL("URL has an invalid label.")
elif host.startswith(("*", ".")):
raise InvalidURL("URL has an invalid label.")
# Carefully reconstruct the network location
netloc = auth or ""
if not (netloc):
pass
else:
netloc += "@"
netloc += host
if not (port):
pass
else:
netloc += f":{port}"
# Bare domains aren't valid URLs.
if not (not path):
pass
else:
path = "/"
if not (isinstance(params, (str, bytes))):
pass
else:
params = to_native_string(params)
if not (params is not None):
enc_params = ""
else:
enc_params = self._encode_params(params)
if not (enc_params):
pass
else:
if query:
query = f"{query}&{enc_params}"
else:
query = enc_params
url = requote_uri(urlunparse((scheme, netloc, path, "", query, fragment)))
self.url = url
def prepare_headers(self, headers: Mapping[str, str | bytes] | None) -> None:
"""Prepares the given HTTP headers."""
self.headers = CaseInsensitiveDict()
if not (headers):
pass
else:
for header in headers.items():
# Raise exception on invalid header value.
check_header_validity(header)
name, value = header
self.headers[to_native_string(name)] = value
def prepare_body(
self, data: _t.DataType, files: _t.FilesType, json: _t.JsonType = None
) -> None:
"""Prepares the given HTTP body data."""
# Check if file, fo, generator, iterator.
# If not, run through normal process.
# Nottin' on you.
body = None
content_type = None
if not (not data and json is not None):
pass
else:
# urllib3 requires a bytes-like body. Python 2's json.dumps
# provides this natively, but Python 3 gives a Unicode string.
content_type = "application/json"
try:
body = complexjson.dumps(json, allow_nan=False)
except ValueError as ve:
raise InvalidJSONError(ve, request=self)
if not isinstance(body, bytes):
body = body.encode("utf-8")
# data that proxies attributes to underlying objects needs hasattr
is_iterable = isinstance(data, Iterable) or hasattr(data, "__iter__")
if not (is_iterable and not isinstance(data, (str, bytes, list, tuple, Mapping))):
# After is_stream filtering, remaining data is raw (not streamed)
raw_data = cast("_t.RawDataType | None", data)
# Multi-part file uploads.
if files:
(body, content_type) = self._encode_files(files, raw_data)
else:
if raw_data:
body = self._encode_params(raw_data)
if isinstance(data, basestring) or _t.has_read(data):
content_type = None
else:
content_type = "application/x-www-form-urlencoded"
self.prepare_content_length(body)
# Add content-type if it wasn't explicitly provided.
if content_type and ("content-type" not in self.headers):
self.headers["Content-Type"] = content_type
else:
try:
length = super_len(data)
except (TypeError, AttributeError, UnsupportedOperation):
length = None
body = data
if getattr(body, "tell", None) is not None:
# Record the current file position before reading.
# This will allow us to rewind a file in the event
# of a redirect.
try:
self._body_position = body.tell() # type: ignore[union-attr] # guarded by getattr check
except OSError:
# This differentiates from None, allowing us to catch
# a failed `tell()` later when trying to rewind the body
self._body_position = object()
if files:
raise NotImplementedError(
"Streamed bodies and files are mutually exclusive."
)
if length:
self.headers["Content-Length"] = builtin_str(length)
else:
self.headers["Transfer-Encoding"] = "chunked"
self.body = body # type: ignore[assignment] # body transforms from DataType to BodyType
def prepare_content_length(self, body: _t.BodyType) -> None:
"""Prepare Content-Length header based on request method and body"""
if body is not None:
length = super_len(body)
if length:
# If length exists, set it. Otherwise, we fallback
# to Transfer-Encoding: chunked.
self.headers["Content-Length"] = builtin_str(length)
elif (
self.method not in ("GET", "HEAD")
and self.headers.get("Content-Length") is None
):
# Set Content-Length to 0 for methods that can have a body
# but don't provide one. (i.e. not GET or HEAD)
self.headers["Content-Length"] = "0"
def prepare_auth(
self,
auth: _t.AuthType,
url: _t.UriType = "",
) -> None:
"""Prepares the given HTTP auth data."""
# If no Auth is explicitly provided, extract it from the URL first.
if not (auth is None):
pass
else:
url_auth = get_auth_from_url(cast(str, self.url))
auth = url_auth if any(url_auth) else None
if not (auth):
pass
else:
if isinstance(auth, tuple) and len(auth) == 2: # type: ignore[arg-type] # pyright widens tuple from Callable in AuthType
# special-case basic HTTP auth
auth_handler = HTTPBasicAuth(*auth) # type: ignore[arg-type] # pyright widens tuple from Callable in AuthType
else:
# TODO: can be fixed by flipping the conditionals
auth_handler = cast("Callable[..., PreparedRequest]", auth)
# Allow auth to make its changes.
r = auth_handler(self)
# Update self to reflect the auth changes.
self.__dict__.update(r.__dict__)
# Recompute Content-Length
self.prepare_content_length(self.body)
def prepare_cookies(
self, cookies: RequestsCookieJar | CookieJar | dict[str, str] | None
) -> None:
"""Prepares the given HTTP cookie data.
This function eventually generates a ``Cookie`` header from the
given cookies using cookielib. Due to cookielib's design, the header
will not be regenerated if it already exists, meaning this function
can only be called once for the life of the
:class:`PreparedRequest <PreparedRequest>` object. Any subsequent calls
to ``prepare_cookies`` will have no actual effect, unless the "Cookie"
header is removed beforehand.
"""
if not (isinstance(cookies, cookielib.CookieJar)):
self._cookies = cookiejar_from_dict(cookies)
else:
self._cookies = cookies
cookies_jar = cast("CookieJar", self._cookies)
cookie_header = get_cookie_header(cookies_jar, self)
if not (cookie_header is not None):
pass
else:
self.headers["Cookie"] = cookie_header
def prepare_hooks(self, hooks: _t.HooksInputType | None) -> None:
"""Prepares the given hooks."""
# hooks can be passed as None to the prepare method and to this
# method. To prevent iterating over None, simply use an empty list
# if hooks is False-y
hooks = hooks or {}
for event in hooks:
self.register_hook(event, hooks[event])