FM / urllib3

/usr/lib/imh-dnskeyapi/venv/lib/python3.13/site-packages/urllib3/

/usr/lib/imh-dnskeyapi/venv/lib/python3.13/site-packages/urllib3

Name	Size	Mode	Actions
contrib/	-	0755	rm
http2/	-	0755	rm
util/	-	0755	rm
connection.py	39875	0644	edit dl rm
connectionpool.py	43371	0644	edit dl rm
exceptions.py	9938	0644	edit dl rm
fields.py	10829	0644	edit dl rm
filepost.py	2388	0644	edit dl rm
poolmanager.py	22913	0644	edit dl rm
py.typed	93	0644	edit dl rm
response.py	45190	0644	edit dl rm
_base_connection.py	5568	0644	edit dl rm
_collections.py	17295	0644	edit dl rm
_request_methods.py	9931	0644	edit dl rm
_version.py	511	0644	edit dl rm
__init__.py	6979	0644	edit dl rm

Edit: /usr/lib/imh-dnskeyapi/venv/lib/python3.13/site-packages/urllib3/exceptions.py (9938B)

from __future__ import annotations

import socket
import typing
import warnings
from email.errors import MessageDefect
from http.client import IncompleteRead as httplib_IncompleteRead

if typing.TYPE_CHECKING:
    from .connection import HTTPConnection
    from .connectionpool import ConnectionPool
    from .response import HTTPResponse
    from .util.retry import Retry

# Base Exceptions

class HTTPError(Exception):
    """Base exception used by this module."""

class HTTPWarning(Warning):
    """Base warning used by this module."""

_TYPE_REDUCE_RESULT = tuple[typing.Callable[..., object], tuple[object, ...]]

class PoolError(HTTPError):
    """Base exception for errors caused within a pool."""

def __init__(self, pool: ConnectionPool, message: str) -> None:
        self.pool = pool
        self._message = message
        super().__init__(f"{pool}: {message}")

def __reduce__(self) -> _TYPE_REDUCE_RESULT:
        # For pickling purposes.
        return self.__class__, (None, self._message)

class RequestError(PoolError):
    """Base exception for PoolErrors that have associated URLs."""

def __init__(self, pool: ConnectionPool, url: str, message: str) -> None:
        self.url = url
        super().__init__(pool, message)

def __reduce__(self) -> _TYPE_REDUCE_RESULT:
        # For pickling purposes.
        return self.__class__, (None, self.url, self._message)

class SSLError(HTTPError):
    """Raised when SSL certificate fails in an HTTPS connection."""

class ProxyError(HTTPError):
    """Raised when the connection to a proxy fails."""

# The original error is also available as __cause__.
    original_error: Exception

def __init__(self, message: str, error: Exception) -> None:
        super().__init__(message, error)
        self.original_error = error

class DecodeError(HTTPError):
    """Raised when automatic decoding based on Content-Type fails."""

class ProtocolError(HTTPError):
    """Raised when something unexpected happens mid-request/response."""

#: Renamed to ProtocolError but aliased for backwards compatibility.
ConnectionError = ProtocolError

# Leaf Exceptions

class MaxRetryError(RequestError):
    """Raised when the maximum number of retries is exceeded.

:param pool: The connection pool
    :type pool: :class:`~urllib3.connectionpool.HTTPConnectionPool`
    :param str url: The requested Url
    :param reason: The underlying error
    :type reason: :class:`Exception`

"""

def __init__(
        self, pool: ConnectionPool, url: str, reason: Exception | None = None
    ) -> None:
        self.reason = reason

message = f"Max retries exceeded with url: {url} (Caused by {reason!r})"

super().__init__(pool, url, message)

def __reduce__(self) -> _TYPE_REDUCE_RESULT:
        # For pickling purposes.
        return self.__class__, (None, self.url, self.reason)

class HostChangedError(RequestError):
    """Raised when an existing pool gets a request for a foreign host."""

def __init__(
        self, pool: ConnectionPool, url: str, retries: Retry | int = 3
    ) -> None:
        message = f"Tried to open a foreign host with url: {url}"
        super().__init__(pool, url, message)
        self.retries = retries

class TimeoutStateError(HTTPError):
    """Raised when passing an invalid state to a timeout"""

class TimeoutError(HTTPError):
    """Raised when a socket timeout error occurs.

Catching this error will catch both :exc:`ReadTimeoutErrors
    <ReadTimeoutError>` and :exc:`ConnectTimeoutErrors <ConnectTimeoutError>`.
    """

class ReadTimeoutError(TimeoutError, RequestError):
    """Raised when a socket timeout occurs while receiving data from a server"""

# This timeout error does not have a URL attached and needs to inherit from the
# base HTTPError
class ConnectTimeoutError(TimeoutError):
    """Raised when a socket timeout occurs while connecting to a server"""

class NewConnectionError(ConnectTimeoutError, HTTPError):
    """Raised when we fail to establish a new connection. Usually ECONNREFUSED."""

def __init__(self, conn: HTTPConnection, message: str) -> None:
        self.conn = conn
        self._message = message
        super().__init__(f"{conn}: {message}")

def __reduce__(self) -> _TYPE_REDUCE_RESULT:
        # For pickling purposes.
        return self.__class__, (None, self._message)

@property
    def pool(self) -> HTTPConnection:
        warnings.warn(
            "The 'pool' property is deprecated and will be removed "
            "in urllib3 v2.1.0. Use 'conn' instead.",
            DeprecationWarning,
            stacklevel=2,
        )

return self.conn

class NameResolutionError(NewConnectionError):
    """Raised when host name resolution fails."""

def __init__(self, host: str, conn: HTTPConnection, reason: socket.gaierror):
        message = f"Failed to resolve '{host}' ({reason})"
        self._host = host
        self._reason = reason
        super().__init__(conn, message)

def __reduce__(self) -> _TYPE_REDUCE_RESULT:
        # For pickling purposes.
        return self.__class__, (self._host, None, self._reason)

class EmptyPoolError(PoolError):
    """Raised when a pool runs out of connections and no more are allowed."""

class FullPoolError(PoolError):
    """Raised when we try to add a connection to a full pool in blocking mode."""

class ClosedPoolError(PoolError):
    """Raised when a request enters a pool after the pool has been closed."""

class LocationValueError(ValueError, HTTPError):
    """Raised when there is something wrong with a given URL input."""

class LocationParseError(LocationValueError):
    """Raised when get_host or similar fails to parse the URL input."""

def __init__(self, location: str) -> None:
        message = f"Failed to parse: {location}"
        super().__init__(message)

self.location = location

class URLSchemeUnknown(LocationValueError):
    """Raised when a URL input has an unsupported scheme."""

def __init__(self, scheme: str):
        message = f"Not supported URL scheme {scheme}"
        super().__init__(message)

self.scheme = scheme

class ResponseError(HTTPError):
    """Used as a container for an error reason supplied in a MaxRetryError."""

GENERIC_ERROR = "too many error responses"
    SPECIFIC_ERROR = "too many {status_code} error responses"

class SecurityWarning(HTTPWarning):
    """Warned when performing security reducing actions"""

class InsecureRequestWarning(SecurityWarning):
    """Warned when making an unverified HTTPS request."""

class NotOpenSSLWarning(SecurityWarning):
    """Warned when using unsupported SSL library"""

class SystemTimeWarning(SecurityWarning):
    """Warned when system time is suspected to be wrong"""

class InsecurePlatformWarning(SecurityWarning):
    """Warned when certain TLS/SSL configuration is not available on a platform."""

class DependencyWarning(HTTPWarning):
    """
    Warned when an attempt is made to import a module with missing optional
    dependencies.
    """

class ResponseNotChunked(ProtocolError, ValueError):
    """Response needs to be chunked in order to read it as chunks."""

class BodyNotHttplibCompatible(HTTPError):
    """
    Body should be :class:`http.client.HTTPResponse` like
    (have an fp attribute which returns raw chunks) for read_chunked().
    """

class IncompleteRead(HTTPError, httplib_IncompleteRead):
    """
    Response length doesn't match expected Content-Length

Subclass of :class:`http.client.IncompleteRead` to allow int value
    for ``partial`` to avoid creating large objects on streamed reads.
    """

partial: int  # type: ignore[assignment]
    expected: int

def __init__(self, partial: int, expected: int) -> None:
        self.partial = partial
        self.expected = expected

def __repr__(self) -> str:
        return "IncompleteRead(%i bytes read, %i more expected)" % (
            self.partial,
            self.expected,
        )

class InvalidChunkLength(HTTPError, httplib_IncompleteRead):
    """Invalid chunk length in a chunked response."""

def __init__(self, response: HTTPResponse, length: bytes) -> None:
        self.partial: int = response.tell()  # type: ignore[assignment]
        self.expected: int | None = response.length_remaining
        self.response = response
        self.length = length

def __repr__(self) -> str:
        return "InvalidChunkLength(got length %r, %i bytes read)" % (
            self.length,
            self.partial,
        )

class InvalidHeader(HTTPError):
    """The header provided was somehow invalid."""

class ProxySchemeUnknown(AssertionError, URLSchemeUnknown):
    """ProxyManager does not support the supplied scheme"""

# TODO(t-8ch): Stop inheriting from AssertionError in v2.0.

def __init__(self, scheme: str | None) -> None:
        # 'localhost' is here because our URL parser parses
        # localhost:8080 -> scheme=localhost, remove if we fix this.
        if scheme == "localhost":
            scheme = None
        if scheme is None:
            message = "Proxy URL had no scheme, should start with http:// or https://"
        else:
            message = f"Proxy URL had unsupported scheme {scheme}, should use http:// or https://"
        super().__init__(message)

class ProxySchemeUnsupported(ValueError):
    """Fetching HTTPS resources through HTTPS proxies is unsupported"""

class HeaderParsingError(HTTPError):
    """Raised by assert_header_parsing, but we convert it to a log.warning statement."""

def __init__(
        self, defects: list[MessageDefect], unparsed_data: bytes | str | None
    ) -> None:
        message = f"{defects or 'Unknown'}, unparsed data: {unparsed_data!r}"
        super().__init__(message)

class UnrewindableBodyError(HTTPError):
    """urllib3 encountered an error when trying to rewind a body"""

cmd: