Add BaseHTTPResponse base class
MODIFY
src/urllib3/connectionpool.py
src/urllib3/connectionpool.py
@@ -34,7 +34,7 @@ from .exceptions import (
)
from .packages.ssl_match_hostname import CertificateError
from .request import RequestMethods
-from .response import HTTPResponse
+from .response import BaseHTTPResponse, HTTPResponse
from .util.connection import is_connection_dropped
from .util.proxy import connection_requires_http_tunnel
from .util.queue import LifoQueue
@@ -484,7 +484,7 @@ class HTTPConnectionPool(ConnectionPool, RequestMethods):
chunked=False,
body_pos=None,
**response_kw,
- ):
+ ) -> BaseHTTPResponse:
"""
Get a connection from the pool and perform an HTTP request. This is the
lowest level call for making a request, so you'll need to specify all

MODIFY
src/urllib3/request.py
src/urllib3/request.py
from urllib.parse import urlencode
from .filepost import encode_multipart_formdata
+from .response import BaseHTTPResponse
__all__ = ["RequestMethods"]
@@ -48,13 +49,15 @@ class RequestMethods:
encode_multipart=True,
multipart_boundary=None,
**kw
- ): # Abstract
+ ) -> BaseHTTPResponse: # Abstract
raise NotImplementedError(
"Classes extending RequestMethods must implement "
"their own ``urlopen`` method."
)
- def request(self, method, url, fields=None, headers=None, **urlopen_kw):
+ def request(
+ self, method, url, fields=None, headers=None, **urlopen_kw
+ ) -> BaseHTTPResponse:
"""
Make a request using :meth:`urlopen` with the appropriate encoding of
``fields`` based on the ``method`` used.
@@ -78,7 +81,9 @@ class RequestMethods:
method, url, fields=fields, headers=headers, **urlopen_kw
)
- def request_encode_url(self, method, url, fields=None, headers=None, **urlopen_kw):
+ def request_encode_url(
+ self, method, url, fields=None, headers=None, **urlopen_kw
+ ) -> BaseHTTPResponse:
"""
Make a request using :meth:`urlopen` with the ``fields`` encoded in
the url. This is useful for request methods like GET, HEAD, DELETE, etc.
@@ -103,7 +108,7 @@ class RequestMethods:
encode_multipart=True,
multipart_boundary=None,
**urlopen_kw
- ):
+ ) -> BaseHTTPResponse:
"""
Make a request using :meth:`urlopen` with the ``fields`` encoded in
the body. This is useful for request methods like POST, PUT, PATCH, etc.

MODIFY
src/urllib3/response.py
src/urllib3/response.py
import io
import logging
+import typing
import zlib
from contextlib import contextmanager
from socket import error as SocketError
@@ -29,15 +30,20 @@ from .util.response import is_fp_closed, is_response_to_head
log = logging.getLogger(__name__)
-class DeflateDecoder:
+class ContentDecoder:
+ def decompress(self, data: bytes) -> bytes:
+ raise NotImplementedError()
+
+ def flush(self) -> bytes:
+ raise NotImplementedError()
+
+
+class DeflateDecoder(ContentDecoder):
def __init__(self):
self._first_try = True
self._data = b""
self._obj = zlib.decompressobj()
- def __getattr__(self, name):
- return getattr(self._obj, name)
-
def decompress(self, data):
if not data:
return data
@@ -60,6 +66,9 @@ class DeflateDecoder:
finally:
self._data = None
+ def flush(self) -> bytes:
+ return self._obj.flush()
+
class GzipDecoderState:
@@ -68,14 +77,11 @@ class GzipDecoderState:
SWALLOW_DATA = 2
-class GzipDecoder:
+class GzipDecoder(ContentDecoder):
def __init__(self):
self._obj = zlib.decompressobj(16 + zlib.MAX_WBITS)
self._state = GzipDecoderState.FIRST_MEMBER
- def __getattr__(self, name):
- return getattr(self._obj, name)
-
def decompress(self, data):
ret = bytearray()
if self._state == GzipDecoderState.SWALLOW_DATA or not data:
@@ -97,10 +103,13 @@ class GzipDecoder:
self._state = GzipDecoderState.OTHER_MEMBERS
self._obj = zlib.decompressobj(16 + zlib.MAX_WBITS)
+ def flush(self) -> bytes:
+ return self._obj.flush()
+
if brotli is not None:
- class BrotliDecoder:
+ class BrotliDecoder(ContentDecoder):
# Supports both 'brotlipy' and 'Brotli' packages
# since they share an import name. The top branches
# are for 'brotlipy' and bottom branches for 'Brotli'
@@ -117,7 +126,7 @@ if brotli is not None:
return b""
-class MultiDecoder:
+class MultiDecoder(ContentDecoder):
"""
From RFC7231:
If one or more encodings have been applied to a representation, the
@@ -138,7 +147,7 @@ class MultiDecoder:
return data
-def _get_decoder(mode):
+def _get_decoder(mode: str) -> ContentDecoder:
if "," in mode:
return MultiDecoder(mode)
@@ -151,7 +160,180 @@ def _get_decoder(mode):
return DeflateDecoder()
-class HTTPResponse(io.IOBase):
+class BaseHTTPResponse(io.IOBase):
+ CONTENT_DECODERS = ["gzip", "deflate"]
+ if brotli is not None:
+ CONTENT_DECODERS += ["br"]
+ REDIRECT_STATUSES = [301, 302, 303, 307, 308]
+
+ DECODER_ERROR_CLASSES = (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,
+ status: int,
+ version: int,
+ reason: str,
+ decode_content: bool,
+ ) -> None:
+ if isinstance(headers, HTTPHeaderDict):
+ self.headers = headers
+ else:
+ self.headers = HTTPHeaderDict(headers)
+ self.status = status
+ self.version = version
+ self.reason = reason
+ self.decode_content = decode_content
+
+ self.chunked = False
+ tr_enc = self.headers.get("transfer-encoding", "").lower()
+ # Don't incur the penalty of creating a list and then discarding it
+ encodings = (enc.strip() for enc in tr_enc.split(","))
+ if "chunked" in encodings:
+ self.chunked = True
+
+ self._decoder: typing.Optional[ContentDecoder] = None
+
+ def get_redirect_location(self) -> typing.Optional[typing.Union[bool, str]]:
+ """
+ Should we redirect and where to?
+
+ :returns: Truthy redirect location string if we got a redirect status
+ code and valid location. ``None`` if redirect status and no
+ location. ``False`` if not a redirect status code.
+ """
+ if self.status in self.REDIRECT_STATUSES:
+ return self.headers.get("location")
+ return False
+
+ @property
+ def data(self) -> bytes:
+ raise NotImplementedError()
+
+ @property
+ def url(self) -> str:
+ raise NotImplementedError()
+
+ @property
+ def closed(self) -> bool:
+ raise NotImplementedError()
+
+ @property
+ def connection(self):
+ raise NotImplementedError()
+
+ def stream(
+ self, amt: int = 2 ** 16, decode_content: typing.Optional[bool] = None
+ ) -> typing.Generator[bytes, None, None]:
+ raise NotImplementedError()
+
+ def read(
+ self,
+ amt: typing.Optional[int] = None,
+ decode_content: typing.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:
+ raise NotImplementedError()
+
+ def release_conn(self) -> None:
+ raise NotImplementedError()
+
+ def drain_conn(self) -> None:
+ raise NotImplementedError()
+
+ def close(self) -> None:
+ raise NotImplementedError()
+
+ def _init_decoder(self) -> None:
+ """
+ Set-up the _decoder attribute if necessary.
+ """
+ # Note: content-encoding value should be case-insensitive, per RFC 7230
+ # Section 3.2
+ content_encoding = self.headers.get("content-encoding", "").lower()
+ if self._decoder is None:
+ if content_encoding in self.CONTENT_DECODERS:
+ self._decoder = _get_decoder(content_encoding)
+ elif "," in content_encoding:
+ encodings = [
+ e.strip()
+ for e in content_encoding.split(",")
+ if e.strip() in self.CONTENT_DECODERS
+ ]
+ if len(encodings):
+ self._decoder = _get_decoder(content_encoding)
+
+ def _decode(self, data: bytes, decode_content: bool, flush_decoder: bool) -> bytes:
+ """
+ Decode the data passed in and potentially flush the decoder.
+ """
+ if not decode_content:
+ return data
+
+ try:
+ if self._decoder:
+ data = self._decoder.decompress(data)
+ except self.DECODER_ERROR_CLASSES as e:
+ content_encoding = self.headers.get("content-encoding", "").lower()
+ raise DecodeError(
+ "Received response with content-encoding: %s, but "
+ "failed to decode it." % content_encoding,
+ e,
+ )
+ if flush_decoder:
+ data += self._flush_decoder()
+
+ return data
+
+ def _flush_decoder(self) -> bytes:
+ """
+ Flushes the decoder. Should only be called if the decoder is actually
+ being used.
+ """
+ if self._decoder:
+ return self._decoder.decompress(b"") + self._decoder.flush()
+ return b""
+
+ # Compatibility methods for `io` module
+ def readable(self) -> bool:
+ return True
+
+ def readinto(self, b: bytearray) -> int:
+ temp = self.read(len(b))
+ if len(temp) == 0:
+ return 0
+ else:
+ b[: len(temp)] = temp
+ return len(temp)
+
+ # Compatibility methods for http.client.HTTPResponse
+ def getheaders(self) -> typing.MutableMapping[str, str]:
+ return self.headers
+
+ def getheader(
+ self, name: str, default: typing.Optional[str] = None
+ ) -> typing.Optional[str]:
+ return self.headers.get(name, default)
+
+ # Compatibility method for http.cookiejar
+ def info(self):
+ return self.headers
+
+ def geturl(self):
+ return self.url
+
+
+class HTTPResponse(BaseHTTPResponse):
"""
HTTP Response container.
@@ -184,11 +366,6 @@ class HTTPResponse(io.IOBase):
value of Content-Length header, if present. Otherwise, raise error.
"""
- CONTENT_DECODERS = ["gzip", "deflate"]
- if brotli is not None:
- CONTENT_DECODERS += ["br"]
- REDIRECT_STATUSES = [301, 302, 303, 307, 308]
-
def __init__(
self,
body="",
@@ -208,20 +385,18 @@ class HTTPResponse(io.IOBase):
request_url=None,
auto_close=True,
):
+ super().__init__(
+ headers=headers,
+ status=status,
+ version=version,
+ reason=reason,
+ decode_content=decode_content,
+ )
- if isinstance(headers, HTTPHeaderDict):
- self.headers = headers
- else:
- self.headers = HTTPHeaderDict(headers)
- self.status = status
- self.version = version
- self.reason = reason
- self.decode_content = decode_content
self.retries = retries
self.enforce_content_length = enforce_content_length
self.auto_close = auto_close
- self._decoder = None
self._body = None
self._fp = None
self._original_response = original_response
@@ -239,13 +414,7 @@ class HTTPResponse(io.IOBase):
self._fp = body
# Are we using the chunked-style of transfer encoding?
- self.chunked = False
self.chunk_left = None
- tr_enc = self.headers.get("transfer-encoding", "").lower()
- # Don't incur the penalty of creating a list and then discarding it
- encodings = (enc.strip() for enc in tr_enc.split(","))
- if "chunked" in encodings:
- self.chunked = True
# Determine length of response
self.length_remaining = self._init_length(request_method)
@@ -254,19 +423,6 @@ class HTTPResponse(io.IOBase):
if preload_content and not self._body:
self._body = self.read(decode_content=decode_content)
- def get_redirect_location(self):
- """
- Should we redirect and where to?
-
- :returns: Truthy redirect location string if we got a redirect status
- code and valid location. ``None`` if redirect status and no
- location. ``False`` if not a redirect status code.
- """
- if self.status in self.REDIRECT_STATUSES:
- return self.headers.get("location")
-
- return False
-
def release_conn(self):
if not self._pool or not self._connection:
return
@@ -361,62 +517,6 @@ class HTTPResponse(io.IOBase):
return length
- def _init_decoder(self):
- """
- Set-up the _decoder attribute if necessary.
- """
- # Note: content-encoding value should be case-insensitive, per RFC 7230
- # Section 3.2
- content_encoding = self.headers.get("content-encoding", "").lower()
- if self._decoder is None:
- if content_encoding in self.CONTENT_DECODERS:
- self._decoder = _get_decoder(content_encoding)
- elif "," in content_encoding:
- encodings = [
- e.strip()
- for e in content_encoding.split(",")
- if e.strip() in self.CONTENT_DECODERS
- ]
- if len(encodings):
- self._decoder = _get_decoder(content_encoding)
-
- DECODER_ERROR_CLASSES = (IOError, zlib.error)
- if brotli is not None:
- DECODER_ERROR_CLASSES += (brotli.error,)
-
- def _decode(self, data, decode_content, flush_decoder):
- """
- Decode the data passed in and potentially flush the decoder.
- """
- if not decode_content:
- return data
-
- try:
- if self._decoder:
- data = self._decoder.decompress(data)
- except self.DECODER_ERROR_CLASSES as e:
- content_encoding = self.headers.get("content-encoding", "").lower()
- raise DecodeError(
- "Received response with content-encoding: %s, but "
- "failed to decode it." % content_encoding,
- e,
- )
- if flush_decoder:
- data += self._flush_decoder()
-
- return data
-
- def _flush_decoder(self):
- """
- Flushes the decoder. Should only be called if the decoder is actually
- being used.
- """
- if self._decoder:
- buf = self._decoder.decompress(b"")
- return buf + self._decoder.flush()
-
- return b""
-
@contextmanager
def _error_catcher(self):
"""
@@ -597,17 +697,6 @@ class HTTPResponse(io.IOBase):
)
return resp
- # Backwards-compatibility methods for http.client.HTTPResponse
- def getheaders(self):
- return self.headers
-
- def getheader(self, name, default=None):
- return self.headers.get(name, default)
-
- # Backwards compatibility for http.cookiejar
- def info(self):
- return self.headers
-
# Overrides from io.IOBase
def close(self):
if not self.closed:
@@ -651,19 +740,6 @@ class HTTPResponse(io.IOBase):
):
return self._fp.flush()
- def readable(self):
- # This method is required for `io` module compatibility.
- return True
-
- def readinto(self, b):
- # This method is required for `io` module compatibility.
- temp = self.read(len(b))
- if len(temp) == 0:
- return 0
- else:
- b[: len(temp)] = temp
- return len(temp)
-
def supports_chunked_reads(self):
"""
Checks if the underlying file-like object looks like a
@@ -779,7 +855,8 @@ class HTTPResponse(io.IOBase):
if self._original_response:
self._original_response.close()
- def geturl(self):
+ @property
+ def url(self) -> str:
"""
Returns the URL that was the source of this response.
If the request that generated this response redirected, this method

Add BaseHTTPResponse base class