Remove stubs for urllib3 (#10812)

Both types-requests and types-influxdb-client now depend on urllib3>=2 instead of types-urllib3. That in turn means that types-caldav, types-slumber and types-requests-oauthlib all depend indirectly on urllib3>=2, since all three stubs packages depend on types-requests.
This commit is contained in:
Alex Waygood
2023-10-01 00:01:12 +01:00
committed by GitHub
parent 460fa14dfc
commit 559d31c4a3
29 changed files with 4 additions and 864 deletions

View File

@@ -74,7 +74,6 @@
"stubs/stripe",
"stubs/tqdm",
"stubs/ttkthemes",
"stubs/urllib3",
"stubs/vobject",
"stubs/WebOb",
"stubs/workalendar",

View File

@@ -1,6 +1,7 @@
version = "1.37.*"
upstream_repository = "https://github.com/influxdata/influxdb-client-python"
requires = ["types-urllib3"]
# requires a version of urllib3 with a py.typed file
requires = ["urllib3>=2"]
[tool.stubtest]
extras = ["extra"]

View File

@@ -1,7 +1,3 @@
# Re-exports from urllib3 (should be fixed in those stubs, not here)
requests\.adapters\.PoolManager\..*
requests\.adapters\.Retry\..*
# Loop variables that leak into the global scope
requests.packages.mod
requests.packages.package

View File

@@ -1,6 +1,7 @@
version = "2.31.*"
upstream_repository = "https://github.com/psf/requests"
requires = ["types-urllib3"]
# requires a version of urllib3 with a py.typed file
requires = ["urllib3>=2"]
[tool.stubtest]
extras = ["socks"]

View File

@@ -1,31 +0,0 @@
urllib3.NullHandler
urllib3._collections.HTTPHeaderDict.from_httplib
urllib3._collections.HTTPHeaderDict.getlist
urllib3._collections.RLock
urllib3.connection.HTTPConnection.request
urllib3.connection.HTTPSConnection.__init__
urllib3.connection.VerifiedHTTPSConnection.__init__
urllib3.connection.VerifiedHTTPSConnection.set_cert
urllib3.connectionpool.ConnectionError
# TODO: remove ResponseCls ignore when https://github.com/python/mypy/issues/13316 is closed
urllib3.connectionpool.HTTPConnectionPool.ResponseCls
urllib3.connectionpool.HTTPConnectionPool.__init__
urllib3.connectionpool.HTTPConnectionPool.urlopen
urllib3.connectionpool.HTTPSConnectionPool.__init__
urllib3.connectionpool.VerifiedHTTPSConnection.__init__
urllib3.connectionpool.VerifiedHTTPSConnection.set_cert
urllib3.poolmanager.PoolManager.connection_from_host
urllib3.poolmanager.PoolManager.connection_from_url
urllib3.poolmanager.PoolManager.urlopen
urllib3.poolmanager.ProxyManager.__init__
urllib3.poolmanager.ProxyManager.connection_from_host
urllib3.poolmanager.ProxyManager.urlopen
urllib3.request.RequestMethods.request_encode_url
urllib3.response.BrotliDecoder
urllib3.util.connection.poll
urllib3.util.connection.select
urllib3.util.ssl_.create_default_context
urllib3.util.ssl_.ssl_wrap_socket
# Metaclass differs:
urllib3.util.retry.Retry

View File

@@ -1,8 +0,0 @@
version = "1.26.*"
upstream_repository = "https://github.com/urllib3/urllib3"
obsolete_since = "2.0.0" # Released on 2023-04-26
partial_stub = true
[tool.stubtest]
ignore_missing_stub = true
extras = ["socks"]

View File

@@ -1,28 +0,0 @@
import logging
from typing import TextIO
from . import connectionpool, filepost, poolmanager, response
from .util import request as _request, retry, timeout, url
__author__: str
__license__: str
__version__: str
HTTPConnectionPool = connectionpool.HTTPConnectionPool
HTTPSConnectionPool = connectionpool.HTTPSConnectionPool
connection_from_url = connectionpool.connection_from_url
encode_multipart_formdata = filepost.encode_multipart_formdata
PoolManager = poolmanager.PoolManager
ProxyManager = poolmanager.ProxyManager
proxy_from_url = poolmanager.proxy_from_url
HTTPResponse = response.HTTPResponse
make_headers = _request.make_headers
get_host = url.get_host
Timeout = timeout.Timeout
Retry = retry.Retry
class NullHandler(logging.Handler):
def emit(self, record): ...
def add_stderr_logger(level: int = 10) -> logging.StreamHandler[TextIO]: ...
def disable_warnings(category: type[Warning] = ...) -> None: ...

View File

@@ -1,55 +0,0 @@
from collections.abc import MutableMapping
from types import TracebackType
from typing import Any, NoReturn, TypeVar
_KT = TypeVar("_KT")
_VT = TypeVar("_VT")
class RLock:
def __enter__(self): ...
def __exit__(
self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None
) -> None: ...
class RecentlyUsedContainer(MutableMapping[_KT, _VT]):
ContainerCls: Any
dispose_func: Any
lock: Any
def __init__(self, maxsize=10, dispose_func=None) -> None: ...
def __getitem__(self, key): ...
def __setitem__(self, key, value) -> None: ...
def __delitem__(self, key) -> None: ...
def __len__(self) -> int: ...
def __iter__(self): ...
def clear(self): ...
def keys(self): ...
class HTTPHeaderDict(MutableMapping[str, str]):
def __init__(self, headers=None, **kwargs) -> None: ...
def __setitem__(self, key, val) -> None: ...
def __getitem__(self, key): ...
def __delitem__(self, key) -> None: ...
def __contains__(self, key): ...
def __eq__(self, other): ...
def __iter__(self) -> NoReturn: ...
def __len__(self) -> int: ...
def __ne__(self, other): ...
values: Any
get: Any
update: Any
iterkeys: Any
itervalues: Any
def pop(self, key, default=...): ...
def discard(self, key): ...
def add(self, key, val): ...
def extend(self, *args, **kwargs): ...
def getlist(self, key): ...
getheaders: Any
getallmatchingheaders: Any
iget: Any
def copy(self): ...
def iteritems(self): ...
def itermerged(self): ...
def items(self): ...
@classmethod
def from_httplib(cls, message, duplicates=...): ...

View File

@@ -1,56 +0,0 @@
import ssl
from builtins import ConnectionError as ConnectionError
from http.client import HTTPConnection as _HTTPConnection, HTTPException as HTTPException
from typing import Any
from . import exceptions, util
from .util import ssl_, ssl_match_hostname
class DummyConnection: ...
BaseSSLError = ssl.SSLError
ConnectTimeoutError = exceptions.ConnectTimeoutError
SystemTimeWarning = exceptions.SystemTimeWarning
match_hostname = ssl_match_hostname.match_hostname
resolve_cert_reqs = ssl_.resolve_cert_reqs
resolve_ssl_version = ssl_.resolve_ssl_version
ssl_wrap_socket = ssl_.ssl_wrap_socket
assert_fingerprint = ssl_.assert_fingerprint
connection = util.connection
port_by_scheme: Any
RECENT_DATE: Any
class HTTPConnection(_HTTPConnection):
default_port: Any
default_socket_options: Any
is_verified: Any
source_address: Any
socket_options: Any
def __init__(self, *args, **kw) -> None: ...
def connect(self): ...
class HTTPSConnection(HTTPConnection):
default_port: Any
key_file: Any
cert_file: Any
def __init__(self, host, port=None, key_file=None, cert_file=None, strict=None, timeout=..., **kw) -> None: ...
sock: Any
def connect(self): ...
class VerifiedHTTPSConnection(HTTPSConnection):
cert_reqs: Any
ca_certs: Any
ssl_version: Any
assert_fingerprint: Any
key_file: Any
cert_file: Any
assert_hostname: Any
def set_cert(
self, key_file=None, cert_file=None, cert_reqs=None, ca_certs=None, assert_hostname=None, assert_fingerprint=None
): ...
sock: Any
auto_open: Any
is_verified: Any
def connect(self): ...

View File

@@ -1,129 +0,0 @@
import queue
from collections.abc import Mapping
from logging import Logger
from types import TracebackType
from typing import Any, ClassVar
from typing_extensions import Literal, Self, TypeAlias
from . import connection, exceptions, request, response
from .connection import BaseSSLError as BaseSSLError, ConnectionError as ConnectionError, HTTPException as HTTPException
from .util import Url, connection as _connection, queue as urllib3queue, retry, ssl_match_hostname, timeout, url
ClosedPoolError = exceptions.ClosedPoolError
ProtocolError = exceptions.ProtocolError
EmptyPoolError = exceptions.EmptyPoolError
HostChangedError = exceptions.HostChangedError
LocationValueError = exceptions.LocationValueError
MaxRetryError = exceptions.MaxRetryError
ProxyError = exceptions.ProxyError
ReadTimeoutError = exceptions.ReadTimeoutError
SSLError = exceptions.SSLError
TimeoutError = exceptions.TimeoutError
InsecureRequestWarning = exceptions.InsecureRequestWarning
CertificateError = ssl_match_hostname.CertificateError
port_by_scheme = connection.port_by_scheme
DummyConnection = connection.DummyConnection
HTTPConnection = connection.HTTPConnection
HTTPSConnection = connection.HTTPSConnection
VerifiedHTTPSConnection = connection.VerifiedHTTPSConnection
RequestMethods = request.RequestMethods
HTTPResponse = response.HTTPResponse
is_connection_dropped = _connection.is_connection_dropped
Retry = retry.Retry
Timeout = timeout.Timeout
get_host = url.get_host
_Timeout: TypeAlias = Timeout | float
_Retries: TypeAlias = Retry | bool | int
xrange: Any
log: Logger
class ConnectionPool:
scheme: ClassVar[str | None]
QueueCls: ClassVar[type[queue.Queue[Any]]]
host: str
port: int | None
def __init__(self, host: str, port: int | None = None) -> None: ...
def __enter__(self) -> Self: ...
def __exit__(
self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
) -> Literal[False]: ...
def close(self) -> None: ...
class HTTPConnectionPool(ConnectionPool, RequestMethods):
scheme: ClassVar[str]
ConnectionCls: ClassVar[type[HTTPConnection | HTTPSConnection]]
ResponseCls: ClassVar[type[HTTPResponse]]
strict: bool
timeout: _Timeout
retries: _Retries | None
pool: urllib3queue.LifoQueue | None
block: bool
proxy: Url | None
proxy_headers: Mapping[str, str]
num_connections: int
num_requests: int
conn_kw: Any
def __init__(
self,
host: str,
port: int | None = None,
strict: bool = False,
timeout: _Timeout = ...,
maxsize: int = 1,
block: bool = False,
headers: Mapping[str, str] | None = None,
retries: _Retries | None = None,
_proxy: Url | None = None,
_proxy_headers: Mapping[str, str] | None = None,
**conn_kw,
) -> None: ...
def close(self) -> None: ...
def is_same_host(self, url: str) -> bool: ...
def urlopen(
self,
method,
url,
body=None,
headers=None,
retries=None,
redirect=True,
assert_same_host=True,
timeout=...,
pool_timeout=None,
release_conn=None,
**response_kw,
): ...
class HTTPSConnectionPool(HTTPConnectionPool):
key_file: str | None
cert_file: str | None
cert_reqs: int | str | None
ca_certs: str | None
ssl_version: int | str | None
assert_hostname: str | Literal[False] | None
assert_fingerprint: str | None
def __init__(
self,
host: str,
port: int | None = None,
strict: bool = False,
timeout: _Timeout = ...,
maxsize: int = 1,
block: bool = False,
headers: Mapping[str, str] | None = None,
retries: _Retries | None = None,
_proxy: Url | None = None,
_proxy_headers: Mapping[str, str] | None = None,
key_file: str | None = None,
cert_file: str | None = None,
cert_reqs: int | str | None = None,
ca_certs: str | None = None,
ssl_version: int | str | None = None,
assert_hostname: str | Literal[False] | None = None,
assert_fingerprint: str | None = None,
**conn_kw,
) -> None: ...
def connection_from_url(url: str, **kw) -> HTTPConnectionPool: ...

View File

@@ -1,45 +0,0 @@
from collections.abc import Mapping
from typing import ClassVar
from typing_extensions import TypedDict
from ..connection import HTTPConnection, HTTPSConnection
from ..connectionpool import HTTPConnectionPool, HTTPSConnectionPool
from ..poolmanager import PoolManager
class _TYPE_SOCKS_OPTIONS(TypedDict):
socks_version: int
proxy_host: str | None
proxy_port: str | None
username: str | None
password: str | None
rdns: bool
class SOCKSConnection(HTTPConnection):
def __init__(self, _socks_options: _TYPE_SOCKS_OPTIONS, *args, **kwargs) -> None: ...
class SOCKSHTTPSConnection(SOCKSConnection, HTTPSConnection): ...
class SOCKSHTTPConnectionPool(HTTPConnectionPool):
ConnectionCls: ClassVar[type[SOCKSConnection]]
class SOCKSHTTPSConnectionPool(HTTPSConnectionPool):
ConnectionCls: ClassVar[type[SOCKSHTTPSConnection]]
class _ConnectionPoolClasses(TypedDict):
http: type[SOCKSHTTPConnectionPool]
https: type[SOCKSHTTPSConnectionPool]
class SOCKSProxyManager(PoolManager):
# has a class-level default, but is overridden on instances, so not a ClassVar
pool_classes_by_scheme: _ConnectionPoolClasses
proxy_url: str
def __init__(
self,
proxy_url: str,
username: str | None = ...,
password: str | None = ...,
num_pools: int = ...,
headers: Mapping[str, str] | None = ...,
**connection_pool_kw,
) -> None: ...

View File

@@ -1,87 +0,0 @@
from email.errors import MessageDefect
from http.client import IncompleteRead as httplib_IncompleteRead
from typing import Any
from urllib3.connectionpool import ConnectionPool, HTTPResponse
from urllib3.util.retry import Retry
class HTTPError(Exception): ...
class HTTPWarning(Warning): ...
class PoolError(HTTPError):
pool: ConnectionPool
def __init__(self, pool: ConnectionPool, message: str) -> None: ...
class RequestError(PoolError):
url: str
def __init__(self, pool: ConnectionPool, url: str, message: str) -> None: ...
class SSLError(HTTPError): ...
class ProxyError(HTTPError):
original_error: Exception
def __init__(self, message: str, error: Exception, *args: Any) -> None: ...
class DecodeError(HTTPError): ...
class ProtocolError(HTTPError): ...
ConnectionError = ProtocolError
class MaxRetryError(RequestError):
reason: Exception | None
def __init__(self, pool: ConnectionPool, url: str, reason: Exception | None = None) -> None: ...
class HostChangedError(RequestError):
retries: Retry | int
def __init__(self, pool: ConnectionPool, url: str, retries: Retry | int = 3) -> None: ...
class TimeoutStateError(HTTPError): ...
class TimeoutError(HTTPError): ...
class ReadTimeoutError(TimeoutError, RequestError): ...
class ConnectTimeoutError(TimeoutError): ...
class NewConnectionError(ConnectTimeoutError, HTTPError): ...
class EmptyPoolError(PoolError): ...
class ClosedPoolError(PoolError): ...
class LocationValueError(ValueError, HTTPError): ...
class LocationParseError(LocationValueError):
location: str
def __init__(self, location: str) -> None: ...
class URLSchemeUnknown(LocationValueError):
scheme: str
def __init__(self, scheme: str) -> None: ...
class ResponseError(HTTPError):
GENERIC_ERROR: str
SPECIFIC_ERROR: str
class SecurityWarning(HTTPWarning): ...
class SubjectAltNameWarning(SecurityWarning): ...
class InsecureRequestWarning(SecurityWarning): ...
class SystemTimeWarning(SecurityWarning): ...
class InsecurePlatformWarning(SecurityWarning): ...
class SNIMissingWarning(HTTPWarning): ...
class DependencyWarning(HTTPWarning): ...
class ResponseNotChunked(ProtocolError, ValueError): ...
class BodyNotHttplibCompatible(HTTPError): ...
class IncompleteRead(HTTPError, httplib_IncompleteRead):
def __init__(self, partial: bytes, expected: int | None) -> None: ...
class InvalidChunkLength(HTTPError, httplib_IncompleteRead):
response: HTTPResponse
length: bytes
def __init__(self, response: HTTPResponse, length: bytes) -> None: ...
class InvalidHeader(HTTPError): ...
class ProxySchemeUnknown(AssertionError, URLSchemeUnknown):
def __init__(self, scheme: str | None) -> None: ...
class ProxySchemeUnsupported(ValueError): ...
class HeaderParsingError(HTTPError):
def __init__(self, defects: list[MessageDefect], unparsed_data: str | bytes | None) -> None: ...
class UnrewindableBodyError(HTTPError): ...

View File

@@ -1,32 +0,0 @@
from collections.abc import Callable, Mapping
from typing import Any
from typing_extensions import TypeAlias
_FieldValue: TypeAlias = str | bytes
_FieldValueTuple: TypeAlias = _FieldValue | tuple[str, _FieldValue] | tuple[str, _FieldValue, str]
def guess_content_type(filename: str | None, default: str = "application/octet-stream") -> str: ...
def format_header_param_rfc2231(name: str, value: _FieldValue) -> str: ...
def format_header_param_html5(name: str, value: _FieldValue) -> str: ...
format_header_param = format_header_param_html5
class RequestField:
data: Any
headers: Any
def __init__(
self,
name: str,
data: _FieldValue,
filename: str | None = None,
headers: Mapping[str, str] | None = None,
header_formatter: Callable[[str, _FieldValue], str] = ...,
) -> None: ...
@classmethod
def from_tuples(
cls, fieldname: str, value: _FieldValueTuple, header_formatter: Callable[[str, _FieldValue], str] = ...
) -> RequestField: ...
def render_headers(self) -> str: ...
def make_multipart(
self, content_disposition: str | None = None, content_type: str | None = None, content_location: str | None = None
) -> None: ...

View File

@@ -1,17 +0,0 @@
from collections.abc import Iterable, Mapping, Sequence
from typing import Any
from typing_extensions import TypeAlias
from . import fields
RequestField = fields.RequestField
writer: Any
_TYPE_FIELDS_SEQUENCE: TypeAlias = Sequence[tuple[str, fields._FieldValueTuple] | RequestField]
_TYPE_FIELDS: TypeAlias = _TYPE_FIELDS_SEQUENCE | Mapping[str, fields._FieldValueTuple]
def choose_boundary() -> str: ...
def iter_field_objects(fields: _TYPE_FIELDS) -> Iterable[RequestField]: ...
def iter_fields(fields): ...
def encode_multipart_formdata(fields: _TYPE_FIELDS, boundary: str | None = None) -> tuple[bytes, str]: ...

View File

@@ -1,32 +0,0 @@
from types import TracebackType
from typing import Any
from typing_extensions import Literal
from .request import RequestMethods
class PoolManager(RequestMethods):
proxy: Any
connection_pool_kw: Any
pools: Any
def __init__(self, num_pools=10, headers=None, **connection_pool_kw) -> None: ...
def __enter__(self): ...
def __exit__(
self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
) -> Literal[False]: ...
def clear(self): ...
def connection_from_host(self, host, port=None, scheme="http"): ...
def connection_from_url(self, url): ...
# TODO: This was the original signature -- copied another one from base class to fix complaint.
# def urlopen(self, method, url, redirect=True, **kw): ...
def urlopen(self, method, url, body=..., headers=..., encode_multipart=..., multipart_boundary=..., **kw): ...
class ProxyManager(PoolManager):
proxy: Any
proxy_headers: Any
def __init__(self, proxy_url, num_pools=10, headers=None, proxy_headers=None, **connection_pool_kw) -> None: ...
def connection_from_host(self, host, port=None, scheme="http"): ...
# TODO: This was the original signature -- copied another one from base class to fix complaint.
# def urlopen(self, method, url, redirect=True, **kw): ...
def urlopen(self, method, url, body=..., headers=..., encode_multipart=..., multipart_boundary=..., **kw): ...
def proxy_from_url(url, **kw): ...

View File

@@ -1,11 +0,0 @@
from typing import Any
class RequestMethods:
headers: Any
def __init__(self, headers=None) -> None: ...
def urlopen(self, method, url, body=None, headers=None, encode_multipart=True, multipart_boundary=None, **kw): ...
def request(self, method, url, fields=None, headers=None, **urlopen_kw): ...
def request_encode_url(self, method, url, fields=None, **urlopen_kw): ...
def request_encode_body(
self, method, url, fields=None, headers=None, encode_multipart=True, multipart_boundary=None, **urlopen_kw
): ...

View File

@@ -1,98 +0,0 @@
import io
from collections.abc import Iterable, Iterator, Mapping
from http.client import HTTPMessage as _HttplibHTTPMessage, HTTPResponse as _HttplibHTTPResponse
from typing import IO, Any
from typing_extensions import Literal, Self, TypeAlias
from urllib3.connectionpool import HTTPConnection
from . import HTTPConnectionPool, Retry
from ._collections import HTTPHeaderDict
_TYPE_BODY: TypeAlias = bytes | IO[Any] | Iterable[bytes] | str
class DeflateDecoder:
def __getattr__(self, name: str) -> Any: ...
def decompress(self, data: bytes) -> bytes: ...
class GzipDecoderState:
FIRST_MEMBER: Literal[0]
OTHER_MEMBERS: Literal[1]
SWALLOW_DATA: Literal[2]
class GzipDecoder:
def __getattr__(self, name: str) -> Any: ...
def decompress(self, data: bytes) -> bytes: ...
# This class is only available if
# `brotli` is available for import.
class BrotliDecoder:
def flush(self) -> bytes: ...
class MultiDecoder:
def __init__(self, modes: str) -> None: ...
def flush(self) -> bytes: ...
def decompress(self, data: bytes) -> bytes: ...
class HTTPResponse(io.IOBase):
CONTENT_DECODERS: list[str]
REDIRECT_STATUSES: list[int]
headers: HTTPHeaderDict
status: int
version: int
reason: str | None
strict: int
decode_content: bool
retries: Retry | None
enforce_content_length: bool
auto_close: bool
msg: _HttplibHTTPMessage | None
chunked: bool
chunk_left: int | None
length_remaining: int | None
def __init__(
self,
body: _TYPE_BODY = "",
headers: Mapping[str, str] | Mapping[bytes, bytes] | None = None,
status: int = 0,
version: int = 0,
reason: str | None = None,
strict: int = 0,
preload_content: bool = True,
decode_content: bool = True,
original_response: _HttplibHTTPResponse | None = None,
pool: HTTPConnectionPool | None = None,
connection: HTTPConnection | None = None,
msg: _HttplibHTTPMessage | None = None,
retries: Retry | None = None,
enforce_content_length: bool = False,
request_method: str | None = None,
request_url: str | None = None,
auto_close: bool = True,
) -> None: ...
def get_redirect_location(self) -> Literal[False] | str | None: ...
def release_conn(self) -> None: ...
def drain_conn(self) -> None: ...
@property
def data(self) -> bytes | Any: ...
@property
def connection(self) -> HTTPConnection | Any: ...
def isclosed(self) -> bool: ...
def tell(self) -> int: ...
def read(self, amt: int | None = None, decode_content: bool | None = None, cache_content: bool = False) -> bytes: ...
def stream(self, amt: int | None = 65536, decode_content: bool | None = None) -> Iterator[bytes]: ...
@classmethod
def from_httplib(cls, r: _HttplibHTTPResponse, **response_kw: Any) -> Self: ...
def getheaders(self) -> HTTPHeaderDict: ...
def getheader(self, name, default=None) -> str | None: ...
def info(self) -> HTTPHeaderDict: ...
def close(self) -> None: ...
@property
def closed(self) -> bool: ...
def fileno(self) -> int: ...
def flush(self) -> None: ...
def readable(self) -> bool: ...
def readinto(self, b: bytearray) -> int: ...
def supports_chunked_reads(self) -> bool: ...
def read_chunked(self, amt: int | None = None, decode_content: bool | None = None) -> Iterator[bytes]: ...
def geturl(self) -> str | None: ...

View File

@@ -1,20 +0,0 @@
import ssl
from . import connection, request, response, retry, ssl_, timeout, url
is_connection_dropped = connection.is_connection_dropped
make_headers = request.make_headers
is_fp_closed = response.is_fp_closed
SSLContext = ssl.SSLContext
HAS_SNI = ssl_.HAS_SNI
assert_fingerprint = ssl_.assert_fingerprint
resolve_cert_reqs = ssl_.resolve_cert_reqs
resolve_ssl_version = ssl_.resolve_ssl_version
ssl_wrap_socket = ssl_.ssl_wrap_socket
current_time = timeout.current_time
Timeout = timeout.Timeout
Retry = retry.Retry
get_host = url.get_host
parse_url = url.parse_url
split_first = url.split_first
Url = url.Url

View File

@@ -1,8 +0,0 @@
from typing import Any
poll: Any
select: Any
HAS_IPV6: bool
def is_connection_dropped(conn): ...
def create_connection(address, timeout=..., source_address=None, socket_options=None): ...

View File

@@ -1,4 +0,0 @@
from queue import Queue
from typing import Any
class LifoQueue(Queue[Any]): ...

View File

@@ -1,11 +0,0 @@
from typing import Any
# from ..packages import six
# b = six.b
ACCEPT_ENCODING: Any
def make_headers(
keep_alive=None, accept_encoding=None, user_agent=None, basic_auth=None, proxy_basic_auth=None, disable_cache=None
): ...

View File

@@ -1 +0,0 @@
def is_fp_closed(obj): ...

View File

@@ -1,84 +0,0 @@
import logging
from collections.abc import Collection
from types import TracebackType
from typing import Any, ClassVar, NamedTuple
from typing_extensions import Literal, Self
from .. import exceptions
from ..connectionpool import ConnectionPool
from ..response import HTTPResponse
ConnectTimeoutError = exceptions.ConnectTimeoutError
MaxRetryError = exceptions.MaxRetryError
ProtocolError = exceptions.ProtocolError
ReadTimeoutError = exceptions.ReadTimeoutError
ResponseError = exceptions.ResponseError
log: logging.Logger
class RequestHistory(NamedTuple):
method: str | None
url: str | None
error: Exception | None
status: int | None
redirect_location: str | None
class Retry:
DEFAULT_ALLOWED_METHODS: ClassVar[frozenset[str]]
RETRY_AFTER_STATUS_CODES: ClassVar[frozenset[int]]
DEFAULT_REMOVE_HEADERS_ON_REDIRECT: ClassVar[frozenset[str]]
DEFAULT_BACKOFF_MAX: ClassVar[int]
total: bool | int | None
connect: int | None
read: int | None
redirect: Literal[True] | int | None
status: int | None
other: int | None
allowed_methods: Collection[str] | Literal[False] | None
status_forcelist: Collection[int]
backoff_factor: float
raise_on_redirect: bool
raise_on_status: bool
history: tuple[RequestHistory, ...]
respect_retry_after_header: bool
remove_headers_on_redirect: frozenset[str]
def __init__(
self,
total: bool | int | None = 10,
connect: int | None = None,
read: int | None = None,
redirect: bool | int | None = None,
status: int | None = None,
other: int | None = None,
allowed_methods: Collection[str] | Literal[False] | None = ...,
status_forcelist: Collection[int] | None = None,
backoff_factor: float = 0,
raise_on_redirect: bool = True,
raise_on_status: bool = True,
history: tuple[RequestHistory, ...] | None = None,
respect_retry_after_header: bool = True,
remove_headers_on_redirect: Collection[str] = ...,
method_whitelist: Collection[str] | None = ...,
) -> None: ...
def new(self, **kw: Any) -> Self: ...
@classmethod
def from_int(
cls, retries: Retry | bool | int | None, redirect: bool | int | None = True, default: Retry | bool | int | None = None
) -> Retry: ...
def get_backoff_time(self) -> float: ...
def parse_retry_after(self, retry_after: str) -> float: ...
def get_retry_after(self, response: HTTPResponse) -> float | None: ...
def sleep_for_retry(self, response: HTTPResponse | None = None) -> bool: ...
def sleep(self, response: HTTPResponse | None = None) -> None: ...
def is_retry(self, method: str, status_code: int, has_retry_after: bool = False) -> bool: ...
def is_exhausted(self) -> bool: ...
def increment(
self,
method: str | None = None,
url: str | None = None,
response: HTTPResponse | None = None,
error: Exception | None = None,
_pool: ConnectionPool | None = None,
_stacktrace: TracebackType | None = None,
) -> Retry: ...

View File

@@ -1,31 +0,0 @@
import ssl
from typing import Any
from .. import exceptions
SSLError = exceptions.SSLError
InsecurePlatformWarning = exceptions.InsecurePlatformWarning
SSLContext = ssl.SSLContext
HAS_SNI: Any
create_default_context: Any
OP_NO_SSLv2: Any
OP_NO_SSLv3: Any
OP_NO_COMPRESSION: Any
DEFAULT_CIPHERS: str
def assert_fingerprint(cert, fingerprint): ...
def resolve_cert_reqs(candidate): ...
def resolve_ssl_version(candidate): ...
def create_urllib3_context(ssl_version=None, cert_reqs=None, options=None, ciphers=None): ...
def ssl_wrap_socket(
sock,
keyfile=None,
certfile=None,
cert_reqs=None,
ca_certs=None,
server_hostname=None,
ssl_version=None,
ciphers=None,
ssl_context=None,
): ...

View File

@@ -1,8 +0,0 @@
from ssl import _PeerCertRetDictType
from typing_extensions import Final
__version__: Final[str]
class CertificateError(ValueError): ...
def match_hostname(cert: _PeerCertRetDictType, hostname: str) -> None: ...

View File

@@ -1,21 +0,0 @@
from typing import Any
from .. import exceptions
TimeoutStateError = exceptions.TimeoutStateError
def current_time(): ...
class Timeout:
DEFAULT_TIMEOUT: Any
total: Any
def __init__(self, total=None, connect=..., read=...) -> None: ...
@classmethod
def from_float(cls, timeout): ...
def clone(self): ...
def start_connect(self): ...
def get_connect_duration(self): ...
@property
def connect_timeout(self): ...
@property
def read_timeout(self): ...

View File

@@ -1,40 +0,0 @@
from typing import NamedTuple
from .. import exceptions
LocationParseError = exceptions.LocationParseError
url_attrs: list[str]
class _UrlBase(NamedTuple):
auth: str | None
fragment: str | None
host: str | None
path: str | None
port: int | None
query: str | None
scheme: str | None
class Url(_UrlBase):
def __new__(
cls,
scheme: str | None = None,
auth: str | None = None,
host: str | None = None,
port: int | None = None,
path: str | None = None,
query: str | None = None,
fragment: str | None = None,
): ...
@property
def hostname(self) -> str | None: ...
@property
def request_uri(self) -> str: ...
@property
def netloc(self) -> str | None: ...
@property
def url(self) -> str: ...
def split_first(s: str, delims: str) -> tuple[str, str, str | None]: ...
def parse_url(url: str) -> Url: ...
def get_host(url: str) -> tuple[str, str | None, str | None]: ...