Annotate urllib3.response (#6871)

stubs/urllib3/@tests/stubtest_allowlist.txt

@@ -1,6 +1,5 @@
 urllib3.HTTPConnectionPool.__init__
 urllib3.HTTPConnectionPool.urlopen
-urllib3.HTTPResponse.__init__
 urllib3.HTTPSConnectionPool.__init__
 urllib3.NullHandler
 urllib3.PoolManager.connection_from_host
@@ -22,7 +21,6 @@ urllib3.connectionpool.ConnectionError
 urllib3.connectionpool.HTTPConnection.request
 urllib3.connectionpool.HTTPConnectionPool.__init__
 urllib3.connectionpool.HTTPConnectionPool.urlopen
-urllib3.connectionpool.HTTPResponse.__init__
 urllib3.connectionpool.HTTPSConnection.__init__
 urllib3.connectionpool.HTTPSConnectionPool.__init__
 urllib3.connectionpool.ProxyError.__init__
@@ -45,10 +43,7 @@ urllib3.poolmanager.ProxyManager.__init__
 urllib3.poolmanager.ProxyManager.connection_from_host
 urllib3.poolmanager.ProxyManager.urlopen
 urllib3.request.RequestMethods.request_encode_url
-urllib3.response.HTTPHeaderDict.from_httplib
-urllib3.response.HTTPHeaderDict.getlist
-urllib3.response.HTTPResponse.__init__
-urllib3.response.PY3
+urllib3.response.BrotliDecoder
 urllib3.util.Retry.is_forced_retry
 urllib3.util.Retry.sleep
 urllib3.util.connection.poll

stubs/urllib3/urllib3/connection.pyi

@@ -1,6 +1,6 @@
 import ssl
 import sys
-from typing import Any
+from typing import IO, Any, Iterable
 
 from . import exceptions, util
 from .packages import ssl_match_hostname
@@ -13,6 +13,8 @@ else:
     from httplib import HTTPConnection as _HTTPConnection, HTTPException as HTTPException
     class ConnectionError(Exception): ...
 
+_TYPE_BODY = bytes | IO[Any] | Iterable[bytes] | str
+
 class DummyConnection: ...
 
 BaseSSLError = ssl.SSLError

stubs/urllib3/urllib3/response.pyi

@@ -1,66 +1,99 @@
 import io
-from typing import Any
+from http.client import HTTPMessage as _HttplibHTTPMessage, HTTPResponse as _HttplibHTTPResponse
+from typing import Any, Iterator, Mapping
+from typing_extensions import Literal
 
-from . import _collections, exceptions
-from .connection import BaseSSLError as BaseSSLError, HTTPException as HTTPException
-from .util import response
+from urllib3.connectionpool import HTTPConnection
 
-HTTPHeaderDict = _collections.HTTPHeaderDict
-ProtocolError = exceptions.ProtocolError
-DecodeError = exceptions.DecodeError
-ReadTimeoutError = exceptions.ReadTimeoutError
-binary_type = bytes  # six.binary_type
-PY3 = True  # six.PY3
-is_fp_closed = response.is_fp_closed
+from . import HTTPConnectionPool, Retry
+from ._collections import HTTPHeaderDict
+from .connection import _TYPE_BODY
 
 class DeflateDecoder:
     def __init__(self) -> None: ...
-    def __getattr__(self, name): ...
-    def decompress(self, data): ...
+    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 __init__(self) -> None: ...
-    def __getattr__(self, name): ...
-    def decompress(self, data): ...
+    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 __init__(self) -> None: ...
+    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: Any
-    REDIRECT_STATUSES: Any
-    headers: Any
-    status: Any
-    version: Any
-    reason: Any
-    strict: Any
-    decode_content: Any
+    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=...,
-        headers=...,
-        status=...,
-        version=...,
-        reason=...,
-        strict=...,
-        preload_content=...,
-        decode_content=...,
-        original_response=...,
-        pool=...,
-        connection=...,
+        body: _TYPE_BODY = ...,
+        headers: Mapping[str, str] | Mapping[bytes, bytes] | None = ...,
+        status: int = ...,
+        version: int = ...,
+        reason: str | None = ...,
+        strict: int = ...,
+        preload_content: bool = ...,
+        decode_content: bool = ...,
+        original_response: _HttplibHTTPResponse | None = ...,
+        pool: HTTPConnectionPool | None = ...,
+        connection: HTTPConnection | None = ...,
+        msg: _HttplibHTTPMessage | None = ...,
+        retries: Retry | None = ...,
+        enforce_content_length: bool = ...,
+        request_method: str | None = ...,
+        request_url: str | None = ...,
+        auto_close: bool = ...,
     ) -> None: ...
-    def get_redirect_location(self): ...
-    def release_conn(self): ...
+    def get_redirect_location(self) -> Literal[False] | str | None: ...
+    def release_conn(self) -> None: ...
+    def drain_conn(self) -> None: ...
     @property
-    def data(self): ...
-    def tell(self): ...
-    def read(self, amt=..., decode_content=..., cache_content=...): ...
-    def stream(self, amt=..., decode_content=...): ...
+    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 = ..., decode_content: bool | None = ..., cache_content: bool = ...) -> bytes: ...
+    def stream(self, amt: int | None = ..., decode_content: bool | None = ...) -> Iterator[bytes]: ...
     @classmethod
-    def from_httplib(cls, r, **response_kw): ...
-    def getheaders(self): ...
-    def getheader(self, name, default=...): ...
-    def close(self): ...
+    def from_httplib(cls, r: _HttplibHTTPResponse, **response_kw: Any) -> HTTPResponse: ...
+    def getheaders(self) -> HTTPHeaderDict: ...
+    def getheader(self, name, default=...) -> str | None: ...
+    def info(self) -> HTTPHeaderDict: ...
+    def close(self) -> None: ...
     @property
-    def closed(self): ...
-    def fileno(self): ...
-    def flush(self): ...
-    def readable(self): ...
-    def readinto(self, b): ...
+    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 = ..., decode_content: bool | None = ...) -> Iterator[bytes]: ...
+    def geturl(self) -> bool | str: ...