Add type hints to urllib3.response
MODIFY
noxfile.py
noxfile.py
@@ -24,6 +24,7 @@ TYPED_FILES = {
"src/urllib3/filepost.py",
"src/urllib3/poolmanager.py",
"src/urllib3/request.py",
+ "src/urllib3/response.py",
"src/urllib3/util/connection.py",
"src/urllib3/util/proxy.py",
"src/urllib3/util/queue.py",

MODIFY
src/urllib3/response.py
src/urllib3/response.py
import io
import logging
-import typing
import zlib
from contextlib import contextmanager
+from http.client import HTTPMessage as _HttplibHTTPMessage
from http.client import HTTPResponse as _HttplibHTTPResponse
from socket import timeout as SocketTimeout
-from typing import Any, Type
+from typing import (
+ TYPE_CHECKING,
+ Any,
+ Generator,
+ Iterator,
+ List,
+ Mapping,
+ Optional,
+ Tuple,
+ Type,
+ Union,
+)
try:
try:
- import brotlicffi as brotli
+ import brotlicffi as brotli # type: ignore
except ImportError:
- import brotli
+ import brotli # type: ignore
except ImportError:
brotli = None
from ._collections import HTTPHeaderDict
-from .connection import BaseSSLError, HTTPException
+from .connection import ( # type: ignore
+ _TYPE_BODY,
+ BaseSSLError,
+ HTTPConnection,
+ HTTPException,
+)
from .exceptions import (
BodyNotHttplibCompatible,
DecodeError,
@@ -30,10 +46,13 @@ from .exceptions import (
SSLError,
)
from .util.response import is_fp_closed, is_response_to_head
+from .util.retry import Retry
-if typing.TYPE_CHECKING:
+if TYPE_CHECKING:
from typing_extensions import Literal
+ from urllib3.connectionpool import HTTPConnectionPool
+
log = logging.getLogger(__name__)
@@ -46,12 +65,12 @@ class ContentDecoder:
class DeflateDecoder(ContentDecoder):
- def __init__(self):
+ def __init__(self) -> None:
self._first_try = True
self._data = b""
self._obj = zlib.decompressobj()
- def decompress(self, data):
+ def decompress(self, data: bytes) -> bytes:
if not data:
return data
@@ -63,7 +82,7 @@ class DeflateDecoder(ContentDecoder):
decompressed = self._obj.decompress(data)
if decompressed:
self._first_try = False
- self._data = None
+ self._data = None # type: ignore
return decompressed
except zlib.error:
self._first_try = False
@@ -71,7 +90,7 @@ class DeflateDecoder(ContentDecoder):
try:
return self.decompress(self._data)
finally:
- self._data = None
+ self._data = None # type: ignore
def flush(self) -> bytes:
return self._obj.flush()
@@ -85,11 +104,11 @@ class GzipDecoderState:
class GzipDecoder(ContentDecoder):
- def __init__(self):
+ def __init__(self) -> None:
self._obj = zlib.decompressobj(16 + zlib.MAX_WBITS)
self._state = GzipDecoderState.FIRST_MEMBER
- def decompress(self, data):
+ def decompress(self, data: bytes) -> bytes:
ret = bytearray()
if self._state == GzipDecoderState.SWALLOW_DATA or not data:
return bytes(ret)
@@ -120,16 +139,16 @@ if brotli is not None:
# Supports both 'brotlipy' and 'Brotli' packages
# since they share an import name. The top branches
# are for 'brotlipy' and bottom branches for 'Brotli'
- def __init__(self):
+ def __init__(self) -> None:
self._obj = brotli.Decompressor()
if hasattr(self._obj, "decompress"):
- self.decompress = self._obj.decompress
+ setattr(self, "decompress", self._obj.decompress)
else:
- self.decompress = self._obj.process
+ setattr(self, "decompress", self._obj.process)
- def flush(self):
+ def flush(self) -> bytes:
if hasattr(self._obj, "flush"):
- return self._obj.flush()
+ return self._obj.flush() # type: ignore
return b""
@@ -142,13 +161,13 @@ class MultiDecoder(ContentDecoder):
they were applied.
"""
- def __init__(self, modes):
+ def __init__(self, modes: str) -> None:
self._decoders = [_get_decoder(m.strip()) for m in modes.split(",")]
- def flush(self):
+ def flush(self) -> bytes:
return self._decoders[0].flush()
- def decompress(self, data):
+ def decompress(self, data: bytes) -> bytes:
for d in reversed(self._decoders):
data = d.decompress(data)
return data
@@ -173,23 +192,23 @@ class BaseHTTPResponse(io.IOBase):
CONTENT_DECODERS += ["br"]
REDIRECT_STATUSES = [301, 302, 303, 307, 308]
- DECODER_ERROR_CLASSES = (IOError, zlib.error)
+ DECODER_ERROR_CLASSES: Tuple[Type[Exception], ...] = (IOError, zlib.error)
if brotli is not None:
DECODER_ERROR_CLASSES += (brotli.error,)
def __init__(
self,
*,
- headers: typing.Optional[typing.Mapping[typing.AnyStr, typing.AnyStr]] = None,
+ headers: Optional[Union[Mapping[str, str], Mapping[bytes, bytes]]] = None,
status: int,
version: int,
- reason: str,
+ reason: Optional[str],
decode_content: bool,
) -> None:
if isinstance(headers, HTTPHeaderDict):
self.headers = headers
else:
- self.headers = HTTPHeaderDict(headers)
+ self.headers = HTTPHeaderDict(headers) # type: ignore
self.status = status
self.version = version
self.reason = reason
@@ -202,9 +221,9 @@ class BaseHTTPResponse(io.IOBase):
if "chunked" in encodings:
self.chunked = True
- self._decoder: typing.Optional[ContentDecoder] = None
+ self._decoder: Optional[ContentDecoder] = None
- def get_redirect_location(self) -> typing.Union[str, "Literal[False]"]:
+ def get_redirect_location(self) -> Union[Optional[str], "Literal[False]"]:
"""
Should we redirect and where to?
@@ -221,7 +240,7 @@ class BaseHTTPResponse(io.IOBase):
raise NotImplementedError()
@property
- def url(self) -> str:
+ def url(self) -> Optional[str]:
raise NotImplementedError()
@property
@@ -229,27 +248,27 @@ class BaseHTTPResponse(io.IOBase):
raise NotImplementedError()
@property
- def connection(self):
+ def connection(self) -> Optional[HTTPConnection]:
raise NotImplementedError()
def stream(
- self, amt: int = 2 ** 16, decode_content: typing.Optional[bool] = None
- ) -> typing.Generator[bytes, None, None]:
+ self, amt: int = 2 ** 16, decode_content: Optional[bool] = None
+ ) -> Iterator[bytes]:
raise NotImplementedError()
def read(
self,
- amt: typing.Optional[int] = None,
- decode_content: typing.Optional[bool] = None,
+ amt: Optional[int] = None,
+ decode_content: Optional[bool] = None,
cache_content: bool = False,
) -> bytes:
raise NotImplementedError()
def read_chunked(
self,
- amt: typing.Optional[int] = None,
- decode_content: typing.Optional[bool] = None,
- ) -> bytes:
+ amt: Optional[int] = None,
+ decode_content: Optional[bool] = None,
+ ) -> Iterator[bytes]:
raise NotImplementedError()
def release_conn(self) -> None:
@@ -280,7 +299,9 @@ class BaseHTTPResponse(io.IOBase):
if encodings:
self._decoder = _get_decoder(content_encoding)
- def _decode(self, data: bytes, decode_content: bool, flush_decoder: bool) -> bytes:
+ def _decode(
+ self, data: bytes, decode_content: Optional[bool], flush_decoder: bool
+ ) -> bytes:
"""
Decode the data passed in and potentially flush the decoder.
"""
@@ -324,19 +345,17 @@ class BaseHTTPResponse(io.IOBase):
return len(temp)
# Compatibility methods for http.client.HTTPResponse
- def getheaders(self) -> typing.List[typing.Tuple[str, str]]:
+ def getheaders(self) -> List[Tuple[str, str]]:
return list(self.headers.items())
- def getheader(
- self, name: str, default: typing.Optional[str] = None
- ) -> typing.Optional[str]:
+ def getheader(self, name: str, default: Optional[str] = None) -> Optional[str]:
return self.headers.get(name, default)
# Compatibility method for http.cookiejar
- def info(self):
+ def info(self) -> HTTPHeaderDict:
return self.headers
- def geturl(self):
+ def geturl(self) -> Optional[Union[str, "Literal[False]"]]:
return self.url
@@ -375,23 +394,23 @@ class HTTPResponse(BaseHTTPResponse):
def __init__(
self,
- body="",
- headers=None,
- status=0,
- version=0,
- reason=None,
- preload_content=True,
- decode_content=True,
- original_response=None,
- pool=None,
- connection=None,
- msg=None,
- retries=None,
- enforce_content_length=False,
- request_method=None,
- request_url=None,
- auto_close=True,
- ):
+ body: _TYPE_BODY = "",
+ headers: Optional[Union[Mapping[str, str], Mapping[bytes, bytes]]] = None,
+ status: int = 0,
+ version: int = 0,
+ reason: Optional[str] = None,
+ preload_content: bool = True,
+ decode_content: bool = True,
+ original_response: Optional[_HttplibHTTPResponse] = None,
+ pool: Optional["HTTPConnectionPool"] = None,
+ connection: Optional[HTTPConnection] = None,
+ msg: Optional[_HttplibHTTPMessage] = None,
+ retries: Optional[Retry] = None,
+ enforce_content_length: bool = False,
+ request_method: Optional[str] = None,
+ request_url: Optional[str] = None,
+ auto_close: bool = True,
+ ) -> None:
super().__init__(
headers=headers,
status=status,
@@ -405,7 +424,7 @@ class HTTPResponse(BaseHTTPResponse):
self.auto_close = auto_close
self._body = None
- self._fp = None
+ self._fp: Optional[_HttplibHTTPResponse] = None
self._original_response = original_response
self._fp_bytes_read = 0
self.msg = msg
@@ -418,10 +437,10 @@ class HTTPResponse(BaseHTTPResponse):
self._connection = connection
if hasattr(body, "read"):
- self._fp = body
+ self._fp = body # type: ignore
# Are we using the chunked-style of transfer encoding?
- self.chunk_left = None
+ self.chunk_left: Optional[int] = None
# Determine length of response
self.length_remaining = self._init_length(request_method)
@@ -430,9 +449,9 @@ class HTTPResponse(BaseHTTPResponse):
if preload_content and not self._body:
self._body = self.read(decode_content=decode_content)
- def release_conn(self):
+ def release_conn(self) -> None:
if not self._pool or not self._connection:
- return
+ return None
self._pool._put_conn(self._connection)
self._connection = None
@@ -449,19 +468,21 @@ class HTTPResponse(BaseHTTPResponse):
pass
@property
- def data(self):
+ def data(self) -> bytes:
# For backwards-compat with earlier urllib3 0.4 and earlier.
if self._body:
- return self._body
+ return self._body # type: ignore
if self._fp:
return self.read(cache_content=True)
+ return None # type: ignore
+
@property
- def connection(self):
+ def connection(self) -> Optional[HTTPConnection]:
return self._connection
- def isclosed(self):
+ def isclosed(self) -> bool:
return is_fp_closed(self._fp)
def tell(self) -> int:
@@ -472,13 +493,14 @@ class HTTPResponse(BaseHTTPResponse):
"""
return self._fp_bytes_read
- def _init_length(self, request_method):
+ def _init_length(self, request_method: Optional[str]) -> Optional[int]:
"""
Set initial length value for Response content if available.
"""
- length = self.headers.get("content-length")
+ length: Optional[int]
+ content_length: Optional[str] = self.headers.get("content-length")
- if length is not None:
+ if content_length is not None:
if self.chunked:
# This Response will fail with an IncompleteRead if it can't be
# received as chunked. This method falls back to attempt reading
@@ -498,11 +520,11 @@ class HTTPResponse(BaseHTTPResponse):
# (e.g. Content-Length: 42, 42). This line ensures the values
# are all valid ints and that as long as the `set` length is 1,
# all values are the same. Otherwise, the header is invalid.
- lengths = {int(val) for val in length.split(",")}
+ lengths = {int(val) for val in content_length.split(",")}
if len(lengths) > 1:
raise InvalidHeader(
"Content-Length contained multiple "
- "unmatching values (%s)" % length
+ "unmatching values (%s)" % content_length
)
length = lengths.pop()
except ValueError:
@@ -511,6 +533,9 @@ class HTTPResponse(BaseHTTPResponse):
if length < 0:
length = None
+ else: # if content_length is None
+ length = None
+
# Convert status to int for comparison
# In some cases, httplib returns a status of "_UNKNOWN"
try:
@@ -525,7 +550,7 @@ class HTTPResponse(BaseHTTPResponse):
return length
@contextmanager
- def _error_catcher(self):
+ def _error_catcher(self) -> Generator[None, None, None]:
"""
Catch low-level python exceptions, instead re-raising urllib3
variants, so that low-level exceptions are not leaked in the
@@ -542,7 +567,7 @@ class HTTPResponse(BaseHTTPResponse):
except SocketTimeout:
# FIXME: Ideally we'd like to include the url in the ReadTimeoutError but
# there is yet no clean way to get at it from this context.
- raise ReadTimeoutError(self._pool, None, "Read timed out.")
+ raise ReadTimeoutError(self._pool, None, "Read timed out.") # type: ignore
except BaseSSLError as e:
# FIXME: Is there a better way to differentiate between SSLErrors?
@@ -550,7 +575,7 @@ class HTTPResponse(BaseHTTPResponse):
# SSL errors related to framing/MAC get wrapped and reraised here
raise SSLError(e)
- raise ReadTimeoutError(self._pool, None, "Read timed out.")
+ raise ReadTimeoutError(self._pool, None, "Read timed out.") # type: ignore
except (HTTPException, OSError) as e:
# This includes IncompleteRead.
@@ -580,7 +605,12 @@ class HTTPResponse(BaseHTTPResponse):
if self._original_response and self._original_response.isclosed():
self.release_conn()
- def read(self, amt=None, decode_content=None, cache_content=False):
+ def read(
+ self,
+ amt: Optional[int] = None,
+ decode_content: Optional[bool] = None,
+ cache_content: bool = False,
+ ) -> bytes:
"""
Similar to :meth:`http.client.HTTPResponse.read`, but with two additional
parameters: ``decode_content`` and ``cache_content``.
@@ -606,7 +636,7 @@ class HTTPResponse(BaseHTTPResponse):
decode_content = self.decode_content
if self._fp is None:
- return
+ return None # type: ignore
flush_decoder = False
fp_closed = getattr(self._fp, "closed", False)
@@ -631,9 +661,10 @@ class HTTPResponse(BaseHTTPResponse):
# no harm in redundantly calling close.
self._fp.close()
flush_decoder = True
- if self.enforce_content_length and self.length_remaining not in (
- 0,
- None,
+ if (
+ self.enforce_content_length
+ and self.length_remaining is not None
+ and self.length_remaining != 0
):
# This is an edge case that httplib failed to cover due
# to concerns of backward compatibility. We're
@@ -654,7 +685,9 @@ class HTTPResponse(BaseHTTPResponse):
return data
- def stream(self, amt=2 ** 16, decode_content=None):
+ def stream(
+ self, amt: int = 2 ** 16, decode_content: Optional[bool] = None
+ ) -> Generator[bytes, None, None]:
"""
A generator wrapper for the read() method. A call will block until
``amt`` bytes have been read from the connection or until the
@@ -693,11 +726,11 @@ class HTTPResponse(BaseHTTPResponse):
headers = r.msg
if not isinstance(headers, HTTPHeaderDict):
- headers = HTTPHeaderDict(headers.items())
+ headers = HTTPHeaderDict(headers.items()) # type: ignore
resp = ResponseCls(
body=r,
- headers=headers,
+ headers=headers, # type: ignore
status=r.status,
version=r.version,
reason=r.reason,
@@ -707,8 +740,8 @@ class HTTPResponse(BaseHTTPResponse):
return resp
# Overrides from io.IOBase
- def close(self):
- if not self.closed:
+ def close(self) -> None:
+ if not self.closed and self._fp:
self._fp.close()
if self._connection:
@@ -718,9 +751,9 @@ class HTTPResponse(BaseHTTPResponse):
io.IOBase.close(self)
@property
- def closed(self):
+ def closed(self) -> bool:
if not self.auto_close:
- return io.IOBase.closed.__get__(self)
+ return io.IOBase.closed.__get__(self) # type: ignore
elif self._fp is None:
return True
elif hasattr(self._fp, "isclosed"):
@@ -730,7 +763,7 @@ class HTTPResponse(BaseHTTPResponse):
else:
return True
- def fileno(self):
+ def fileno(self) -> int:
if self._fp is None:
raise OSError("HTTPResponse has no file to get a fileno from")
elif hasattr(self._fp, "fileno"):
@@ -741,7 +774,7 @@ class HTTPResponse(BaseHTTPResponse):
"around has no file descriptor"
)
- def flush(self):
+ def flush(self) -> None:
if (
self._fp is not None
and hasattr(self._fp, "flush")
@@ -749,7 +782,7 @@ class HTTPResponse(BaseHTTPResponse):
):
return self._fp.flush()
- def supports_chunked_reads(self):
+ def supports_chunked_reads(self) -> bool:
"""
Checks if the underlying file-like object looks like a
:class:`http.client.HTTPResponse` object. We do this by testing for
@@ -758,12 +791,12 @@ class HTTPResponse(BaseHTTPResponse):
"""
return hasattr(self._fp, "fp")
- def _update_chunk_length(self):
+ def _update_chunk_length(self) -> None:
# First, we'll figure out length of a chunk and then
# we'll try to read it from socket.
if self.chunk_left is not None:
- return
- line = self._fp.fp.readline()
+ return None
+ line = self._fp.fp.readline() # type: ignore
line = line.split(b";", 1)[0]
try:
self.chunk_left = int(line, 16)
@@ -772,29 +805,31 @@ class HTTPResponse(BaseHTTPResponse):
self.close()
raise InvalidChunkLength(self, line)
- def _handle_chunk(self, amt):
+ def _handle_chunk(self, amt: Optional[int]) -> bytes:
returned_chunk = None
if amt is None:
- chunk = self._fp._safe_read(self.chunk_left)
+ chunk = self._fp._safe_read(self.chunk_left) # type: ignore
returned_chunk = chunk
- self._fp._safe_read(2) # Toss the CRLF at the end of the chunk.
+ self._fp._safe_read(2) # type: ignore # Toss the CRLF at the end of the chunk.
self.chunk_left = None
- elif amt < self.chunk_left:
- value = self._fp._safe_read(amt)
+ elif self.chunk_left is not None and amt < self.chunk_left:
+ value = self._fp._safe_read(amt) # type: ignore
self.chunk_left = self.chunk_left - amt
returned_chunk = value
elif amt == self.chunk_left:
- value = self._fp._safe_read(amt)
- self._fp._safe_read(2) # Toss the CRLF at the end of the chunk.
+ value = self._fp._safe_read(amt) # type: ignore
+ self._fp._safe_read(2) # type: ignore # Toss the CRLF at the end of the chunk.
self.chunk_left = None
returned_chunk = value
else: # amt > self.chunk_left
- returned_chunk = self._fp._safe_read(self.chunk_left)
- self._fp._safe_read(2) # Toss the CRLF at the end of the chunk.
+ returned_chunk = self._fp._safe_read(self.chunk_left) # type: ignore
+ self._fp._safe_read(2) # type: ignore # Toss the CRLF at the end of the chunk.
self.chunk_left = None
- return returned_chunk
+ return returned_chunk # type: ignore
- def read_chunked(self, amt=None, decode_content=None):
+ def read_chunked(
+ self, amt: Optional[int] = None, decode_content: Optional[bool] = None
+ ) -> Generator[bytes, None, None]:
"""
Similar to :meth:`HTTPResponse.read`, but with an additional
parameter: ``decode_content``.
@@ -825,12 +860,12 @@ class HTTPResponse(BaseHTTPResponse):
# Don't bother reading the body of a HEAD request.
if self._original_response and is_response_to_head(self._original_response):
self._original_response.close()
- return
+ return None
# If a response is already read and closed
# then return immediately.
- if self._fp.fp is None:
- return
+ if self._fp.fp is None: # type: ignore
+ return None
while True:
self._update_chunk_length()
@@ -852,8 +887,8 @@ class HTTPResponse(BaseHTTPResponse):
yield decoded
# Chunk content ends with \r\n: discard it.
- while True:
- line = self._fp.fp.readline()
+ while self._fp is not None:
+ line = self._fp.fp.readline() # type: ignore
if not line:
# Some sites may not end with '\r\n'.
break
@@ -865,7 +900,7 @@ class HTTPResponse(BaseHTTPResponse):
self._original_response.close()
@property
- def url(self) -> str:
+ def url(self) -> Optional[str]:
"""
Returns the URL that was the source of this response.
If the request that generated this response redirected, this method
@@ -876,16 +911,16 @@ class HTTPResponse(BaseHTTPResponse):
else:
return self._request_url
- def __iter__(self):
- buffer = []
+ def __iter__(self) -> Iterator[bytes]:
+ buffer: List[bytes] = []
for chunk in self.stream(decode_content=True):
if b"\n" in chunk:
- chunk = chunk.split(b"\n")
- yield b"".join(buffer) + chunk[0] + b"\n"
- for x in chunk[1:-1]:
+ chunks = chunk.split(b"\n")
+ yield b"".join(buffer) + chunks[0] + b"\n"
+ for x in chunks[1:-1]:
yield x + b"\n"
- if chunk[-1]:
- buffer = [chunk[-1]]
+ if chunks[-1]:
+ buffer = [chunks[-1]]
else:
buffer = []
else:

MODIFY
src/urllib3/util/retry.py
src/urllib3/util/retry.py
@@ -2,10 +2,9 @@ import email
import logging
import re
import time
-from collections import namedtuple
from itertools import takewhile
from types import TracebackType
-from typing import TYPE_CHECKING, Any, Collection, Optional, Tuple, Union
+from typing import TYPE_CHECKING, Any, Collection, NamedTuple, Optional, Tuple, Union
from ..exceptions import (
ConnectTimeoutError,
@@ -26,9 +25,12 @@ log = logging.getLogger(__name__)
# Data structure for representing the metadata of requests that result in a retry.
-RequestHistory = namedtuple(
- "RequestHistory", ["method", "url", "error", "status", "redirect_location"]
-)
+class RequestHistory(NamedTuple):
+ method: Optional[str]
+ url: Optional[str]
+ error: Optional[Exception]
+ status: Optional[int]
+ redirect_location: Optional[str]
class Retry:
@@ -461,7 +463,9 @@ class Retry:
if redirect is not None:
redirect -= 1
cause = "too many redirects"
- redirect_location = response.get_redirect_location()
+ response_redirect_location = response.get_redirect_location()
+ if response_redirect_location:
+ redirect_location = response_redirect_location
status = response.status
else:

Add type hints to urllib3.response