D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
proc
/
self
/
root
/
opt
/
cloudlinux
/
venv
/
lib
/
python3.11
/
site-packages
/
urllib3
/
util
/
Filename :
request.py
back
Copy
from __future__ import annotations import io import typing from base64 import b64encode from enum import Enum from ..exceptions import UnrewindableBodyError from .util import to_bytes if typing.TYPE_CHECKING: from typing_extensions import Final # 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: try: import brotlicffi as _unused_module_brotli # type: ignore[import] # noqa: F401 except ImportError: import brotli as _unused_module_brotli # type: ignore[import] # noqa: F401 except ImportError: pass else: ACCEPT_ENCODING += ",br" try: import zstandard as _unused_module_zstd # type: ignore[import] # noqa: F401 except ImportError: pass else: ACCEPT_ENCODING += ",zstd" class _TYPE_FAILEDTELL(Enum): token = 0 _FAILEDTELL: Final[_TYPE_FAILEDTELL] = _TYPE_FAILEDTELL.token _TYPE_BODY_POSITION = typing.Union[int, _TYPE_FAILEDTELL] # When sending a request with these methods we aren't expecting # a body so don't need to set an explicit 'Content-Length: 0' # The reason we do this in the negative instead of tracking methods # which 'should' have a body is because unknown methods should be # treated as if they were 'POST' which *does* expect a body. _METHODS_NOT_EXPECTING_BODY = {"GET", "HEAD", "DELETE", "TRACE", "OPTIONS", "CONNECT"} def make_headers( keep_alive: bool | None = None, accept_encoding: bool | list[str] | str | None = None, user_agent: str | None = None, basic_auth: str | None = None, proxy_basic_auth: str | None = None, disable_cache: bool | None = None, ) -> dict[str, str]: """ 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'. If either the ``brotli`` or ``brotlicffi`` package is installed 'gzip,deflate,br' is used instead. 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: .. code-block:: python import urllib3 print(urllib3.util.make_headers(keep_alive=True, user_agent="Batman/1.0")) # {'connection': 'keep-alive', 'user-agent': 'Batman/1.0'} print(urllib3.util.make_headers(accept_encoding=True)) # {'accept-encoding': 'gzip,deflate'} """ headers: dict[str, str] = {} 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" ] = f"Basic {b64encode(basic_auth.encode('latin-1')).decode()}" if proxy_basic_auth: headers[ "proxy-authorization" ] = f"Basic {b64encode(proxy_basic_auth.encode('latin-1')).decode()}" if disable_cache: headers["cache-control"] = "no-cache" return headers def set_file_position( body: typing.Any, pos: _TYPE_BODY_POSITION | None ) -> _TYPE_BODY_POSITION | None: """ 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 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: typing.IO[typing.AnyStr], body_pos: _TYPE_BODY_POSITION) -> None: """ 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, int): try: body_seek(body_pos) except OSError as e: raise UnrewindableBodyError( "An error occurred when rewinding request body for redirect/retry." ) from e elif body_pos is _FAILEDTELL: raise UnrewindableBodyError( "Unable to record file position for rewinding " "request body during a redirect/retry." ) else: raise ValueError( f"body_pos must be of type integer, instead it was {type(body_pos)}." ) class ChunksAndContentLength(typing.NamedTuple): chunks: typing.Iterable[bytes] | None content_length: int | None def body_to_chunks( body: typing.Any | None, method: str, blocksize: int ) -> ChunksAndContentLength: """Takes the HTTP request method, body, and blocksize and transforms them into an iterable of chunks to pass to socket.sendall() and an optional 'Content-Length' header. A 'Content-Length' of 'None' indicates the length of the body can't be determined so should use 'Transfer-Encoding: chunked' for framing instead. """ chunks: typing.Iterable[bytes] | None content_length: int | None # No body, we need to make a recommendation on 'Content-Length' # based on whether that request method is expected to have # a body or not. if body is None: chunks = None if method.upper() not in _METHODS_NOT_EXPECTING_BODY: content_length = 0 else: content_length = None # Bytes or strings become bytes elif isinstance(body, (str, bytes)): chunks = (to_bytes(body),) content_length = len(chunks[0]) # File-like object, TODO: use seek() and tell() for length? elif hasattr(body, "read"): def chunk_readable() -> typing.Iterable[bytes]: nonlocal body, blocksize encode = isinstance(body, io.TextIOBase) while True: datablock = body.read(blocksize) if not datablock: break if encode: datablock = datablock.encode("iso-8859-1") yield datablock chunks = chunk_readable() content_length = None # Otherwise we need to start checking via duck-typing. else: try: # Check if the body implements the buffer API. mv = memoryview(body) except TypeError: try: # Check if the body is an iterable chunks = iter(body) content_length = None except TypeError: raise TypeError( f"'body' must be a bytes-like object, file-like " f"object, or iterable. Instead was {body!r}" ) from None else: # Since it implements the buffer API can be passed directly to socket.sendall() chunks = (body,) content_length = mv.nbytes return ChunksAndContentLength(chunks=chunks, content_length=content_length)