Source code for urllib3.util.timeout

from __future__ import annotations

import time
import typing
from enum import Enum
from socket import getdefaulttimeout

from ..exceptions import TimeoutStateError

if typing.TYPE_CHECKING:
    from typing import Final

class _TYPE_DEFAULT(Enum):
    # This value should never be passed to socket.settimeout() so for safety we use a -1.
    # socket.settimout() raises a ValueError for negative values.
    token = -1


_TYPE_TIMEOUT = typing.Optional[typing.Union[float, _TYPE_DEFAULT]]

[docs] class Timeout: """Timeout configuration. Timeouts can be defined as a default for a pool: .. code-block:: python import urllib3 timeout = urllib3.util.Timeout(connect=2.0, read=7.0) http = urllib3.PoolManager(timeout=timeout) resp = http.request("GET", "") print(resp.status) Or per-request (which overrides the default for the pool): .. code-block:: python response = http.request("GET", "", timeout=Timeout(10)) Timeouts can be disabled by setting all the parameters to ``None``: .. code-block:: python no_timeout = Timeout(connect=None, read=None) response = http.request("GET", "", timeout=no_timeout) :param total: This combines the connect and read timeouts into one; the read timeout will be set to the time leftover from the connect attempt. In the event that both a connect timeout and a total are specified, or a read timeout and a total are specified, the shorter timeout will be applied. Defaults to None. :type total: int, float, or None :param connect: The maximum amount of time (in seconds) to wait for a connection attempt to a server to succeed. Omitting the parameter will default the connect timeout to the system default, probably `the global default timeout in <>`_. None will set an infinite timeout for connection attempts. :type connect: int, float, or None :param read: The maximum amount of time (in seconds) to wait between consecutive read operations for a response from the server. Omitting the parameter will default the read timeout to the system default, probably `the global default timeout in <>`_. None will set an infinite timeout. :type read: int, float, or None .. note:: Many factors can affect the total amount of time for urllib3 to return an HTTP response. For example, Python's DNS resolver does not obey the timeout specified on the socket. Other factors that can affect total request time include high CPU load, high swap, the program running at a low priority level, or other behaviors. In addition, the read and total timeouts only measure the time between read operations on the socket connecting the client and the server, not the total amount of time for the request to return a complete response. For most requests, the timeout is raised because the server has not sent the first byte in the specified time. This is not always the case; if a server streams one byte every fifteen seconds, a timeout of 20 seconds will not trigger, even though the request will take several minutes to complete. """ #: A sentinel object representing the default timeout value DEFAULT_TIMEOUT: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT def __init__( self, total: _TYPE_TIMEOUT = None, connect: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT, read: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT, ) -> None: self._connect = self._validate_timeout(connect, "connect") self._read = self._validate_timeout(read, "read") = self._validate_timeout(total, "total") self._start_connect: float | None = None def __repr__(self) -> str: return f"{type(self).__name__}(connect={self._connect!r}, read={self._read!r}, total={!r})" # __str__ provided for backwards compatibility __str__ = __repr__ @staticmethod def resolve_default_timeout(timeout: _TYPE_TIMEOUT) -> float | None: return getdefaulttimeout() if timeout is _DEFAULT_TIMEOUT else timeout @classmethod def _validate_timeout(cls, value: _TYPE_TIMEOUT, name: str) -> _TYPE_TIMEOUT: """Check that a timeout attribute is valid. :param value: The timeout value to validate :param name: The name of the timeout attribute to validate. This is used to specify in error messages. :return: The validated and casted version of the given value. :raises ValueError: If it is a numeric value less than or equal to zero, or the type is not an integer, float, or None. """ if value is None or value is _DEFAULT_TIMEOUT: return value if isinstance(value, bool): raise ValueError( "Timeout cannot be a boolean value. It must " "be an int, float or None." ) try: float(value) except (TypeError, ValueError): raise ValueError( "Timeout value %s was %s, but it must be an " "int, float or None." % (name, value) ) from None try: if value <= 0: raise ValueError( "Attempted to set %s timeout to %s, but the " "timeout cannot be set to a value less " "than or equal to 0." % (name, value) ) except TypeError: raise ValueError( "Timeout value %s was %s, but it must be an " "int, float or None." % (name, value) ) from None return value
[docs] @classmethod def from_float(cls, timeout: _TYPE_TIMEOUT) -> Timeout: """Create a new Timeout from a legacy timeout value. The timeout value used by sets the same timeout on the connect(), and recv() socket requests. This creates a :class:`Timeout` object that sets the individual timeouts to the ``timeout`` value passed to this function. :param timeout: The legacy timeout value. :type timeout: integer, float, :attr:`urllib3.util.Timeout.DEFAULT_TIMEOUT`, or None :return: Timeout object :rtype: :class:`Timeout` """ return Timeout(read=timeout, connect=timeout)
[docs] def clone(self) -> Timeout: """Create a copy of the timeout object Timeout properties are stored per-pool but each request needs a fresh Timeout object to ensure each one has its own start/stop configured. :return: a copy of the timeout object :rtype: :class:`Timeout` """ # We can't use copy.deepcopy because that will also create a new object # for _GLOBAL_DEFAULT_TIMEOUT, which uses as a sentinel to # detect the user default. return Timeout(connect=self._connect, read=self._read,
[docs] def start_connect(self) -> float: """Start the timeout clock, used during a connect() attempt :raises urllib3.exceptions.TimeoutStateError: if you attempt to start a timer that has been started already. """ if self._start_connect is not None: raise TimeoutStateError("Timeout timer has already been started.") self._start_connect = time.monotonic() return self._start_connect
[docs] def get_connect_duration(self) -> float: """Gets the time elapsed since the call to :meth:`start_connect`. :return: Elapsed time in seconds. :rtype: float :raises urllib3.exceptions.TimeoutStateError: if you attempt to get duration for a timer that hasn't been started. """ if self._start_connect is None: raise TimeoutStateError( "Can't get connect duration for timer that has not started." ) return time.monotonic() - self._start_connect
@property def connect_timeout(self) -> _TYPE_TIMEOUT: """Get the value to use when setting a connection timeout. This will be a positive float or integer, the value None (never timeout), or the default system timeout. :return: Connect timeout. :rtype: int, float, :attr:`Timeout.DEFAULT_TIMEOUT` or None """ if is None: return self._connect if self._connect is None or self._connect is _DEFAULT_TIMEOUT: return return min(self._connect, # type: ignore[type-var] @property def read_timeout(self) -> float | None: """Get the value for the read timeout. This assumes some time has elapsed in the connection timeout and computes the read timeout appropriately. If is set, the read timeout is dependent on the amount of time taken by the connect timeout. If the connection time has not been established, a :exc:`~urllib3.exceptions.TimeoutStateError` will be raised. :return: Value to use for the read timeout. :rtype: int, float or None :raises urllib3.exceptions.TimeoutStateError: If :meth:`start_connect` has not yet been called on this object. """ if ( is not None and is not _DEFAULT_TIMEOUT and self._read is not None and self._read is not _DEFAULT_TIMEOUT ): # In case the connect timeout has not yet been established. if self._start_connect is None: return self._read return max(0, min( - self.get_connect_duration(), self._read)) elif is not None and is not _DEFAULT_TIMEOUT: return max(0, - self.get_connect_duration()) else: return self.resolve_default_timeout(self._read)