jpayne@7: from __future__ import annotations
jpayne@7: 
jpayne@7: import collections
jpayne@7: import io
jpayne@7: import json as _json
jpayne@7: import logging
jpayne@7: import re
jpayne@7: import sys
jpayne@7: import typing
jpayne@7: import warnings
jpayne@7: import zlib
jpayne@7: from contextlib import contextmanager
jpayne@7: from http.client import HTTPMessage as _HttplibHTTPMessage
jpayne@7: from http.client import HTTPResponse as _HttplibHTTPResponse
jpayne@7: from socket import timeout as SocketTimeout
jpayne@7: 
jpayne@7: if typing.TYPE_CHECKING:
jpayne@7:     from ._base_connection import BaseHTTPConnection
jpayne@7: 
jpayne@7: try:
jpayne@7:     try:
jpayne@7:         import brotlicffi as brotli  # type: ignore[import-not-found]
jpayne@7:     except ImportError:
jpayne@7:         import brotli  # type: ignore[import-not-found]
jpayne@7: except ImportError:
jpayne@7:     brotli = None
jpayne@7: 
jpayne@7: try:
jpayne@7:     import zstandard as zstd  # type: ignore[import-not-found]
jpayne@7: 
jpayne@7:     # The package 'zstandard' added the 'eof' property starting
jpayne@7:     # in v0.18.0 which we require to ensure a complete and
jpayne@7:     # valid zstd stream was fed into the ZstdDecoder.
jpayne@7:     # See: https://github.com/urllib3/urllib3/pull/2624
jpayne@7:     _zstd_version = _zstd_version = tuple(
jpayne@7:         map(int, re.search(r"^([0-9]+)\.([0-9]+)", zstd.__version__).groups())  # type: ignore[union-attr]
jpayne@7:     )
jpayne@7:     if _zstd_version < (0, 18):  # Defensive:
jpayne@7:         zstd = None
jpayne@7: 
jpayne@7: except (AttributeError, ImportError, ValueError):  # Defensive:
jpayne@7:     zstd = None
jpayne@7: 
jpayne@7: from . import util
jpayne@7: from ._base_connection import _TYPE_BODY
jpayne@7: from ._collections import HTTPHeaderDict
jpayne@7: from .connection import BaseSSLError, HTTPConnection, HTTPException
jpayne@7: from .exceptions import (
jpayne@7:     BodyNotHttplibCompatible,
jpayne@7:     DecodeError,
jpayne@7:     HTTPError,
jpayne@7:     IncompleteRead,
jpayne@7:     InvalidChunkLength,
jpayne@7:     InvalidHeader,
jpayne@7:     ProtocolError,
jpayne@7:     ReadTimeoutError,
jpayne@7:     ResponseNotChunked,
jpayne@7:     SSLError,
jpayne@7: )
jpayne@7: from .util.response import is_fp_closed, is_response_to_head
jpayne@7: from .util.retry import Retry
jpayne@7: 
jpayne@7: if typing.TYPE_CHECKING:
jpayne@7:     from typing import Literal
jpayne@7: 
jpayne@7:     from .connectionpool import HTTPConnectionPool
jpayne@7: 
jpayne@7: log = logging.getLogger(__name__)
jpayne@7: 
jpayne@7: 
jpayne@7: class ContentDecoder:
jpayne@7:     def decompress(self, data: bytes) -> bytes:
jpayne@7:         raise NotImplementedError()
jpayne@7: 
jpayne@7:     def flush(self) -> bytes:
jpayne@7:         raise NotImplementedError()
jpayne@7: 
jpayne@7: 
jpayne@7: class DeflateDecoder(ContentDecoder):
jpayne@7:     def __init__(self) -> None:
jpayne@7:         self._first_try = True
jpayne@7:         self._data = b""
jpayne@7:         self._obj = zlib.decompressobj()
jpayne@7: 
jpayne@7:     def decompress(self, data: bytes) -> bytes:
jpayne@7:         if not data:
jpayne@7:             return data
jpayne@7: 
jpayne@7:         if not self._first_try:
jpayne@7:             return self._obj.decompress(data)
jpayne@7: 
jpayne@7:         self._data += data
jpayne@7:         try:
jpayne@7:             decompressed = self._obj.decompress(data)
jpayne@7:             if decompressed:
jpayne@7:                 self._first_try = False
jpayne@7:                 self._data = None  # type: ignore[assignment]
jpayne@7:             return decompressed
jpayne@7:         except zlib.error:
jpayne@7:             self._first_try = False
jpayne@7:             self._obj = zlib.decompressobj(-zlib.MAX_WBITS)
jpayne@7:             try:
jpayne@7:                 return self.decompress(self._data)
jpayne@7:             finally:
jpayne@7:                 self._data = None  # type: ignore[assignment]
jpayne@7: 
jpayne@7:     def flush(self) -> bytes:
jpayne@7:         return self._obj.flush()
jpayne@7: 
jpayne@7: 
jpayne@7: class GzipDecoderState:
jpayne@7:     FIRST_MEMBER = 0
jpayne@7:     OTHER_MEMBERS = 1
jpayne@7:     SWALLOW_DATA = 2
jpayne@7: 
jpayne@7: 
jpayne@7: class GzipDecoder(ContentDecoder):
jpayne@7:     def __init__(self) -> None:
jpayne@7:         self._obj = zlib.decompressobj(16 + zlib.MAX_WBITS)
jpayne@7:         self._state = GzipDecoderState.FIRST_MEMBER
jpayne@7: 
jpayne@7:     def decompress(self, data: bytes) -> bytes:
jpayne@7:         ret = bytearray()
jpayne@7:         if self._state == GzipDecoderState.SWALLOW_DATA or not data:
jpayne@7:             return bytes(ret)
jpayne@7:         while True:
jpayne@7:             try:
jpayne@7:                 ret += self._obj.decompress(data)
jpayne@7:             except zlib.error:
jpayne@7:                 previous_state = self._state
jpayne@7:                 # Ignore data after the first error
jpayne@7:                 self._state = GzipDecoderState.SWALLOW_DATA
jpayne@7:                 if previous_state == GzipDecoderState.OTHER_MEMBERS:
jpayne@7:                     # Allow trailing garbage acceptable in other gzip clients
jpayne@7:                     return bytes(ret)
jpayne@7:                 raise
jpayne@7:             data = self._obj.unused_data
jpayne@7:             if not data:
jpayne@7:                 return bytes(ret)
jpayne@7:             self._state = GzipDecoderState.OTHER_MEMBERS
jpayne@7:             self._obj = zlib.decompressobj(16 + zlib.MAX_WBITS)
jpayne@7: 
jpayne@7:     def flush(self) -> bytes:
jpayne@7:         return self._obj.flush()
jpayne@7: 
jpayne@7: 
jpayne@7: if brotli is not None:
jpayne@7: 
jpayne@7:     class BrotliDecoder(ContentDecoder):
jpayne@7:         # Supports both 'brotlipy' and 'Brotli' packages
jpayne@7:         # since they share an import name. The top branches
jpayne@7:         # are for 'brotlipy' and bottom branches for 'Brotli'
jpayne@7:         def __init__(self) -> None:
jpayne@7:             self._obj = brotli.Decompressor()
jpayne@7:             if hasattr(self._obj, "decompress"):
jpayne@7:                 setattr(self, "decompress", self._obj.decompress)
jpayne@7:             else:
jpayne@7:                 setattr(self, "decompress", self._obj.process)
jpayne@7: 
jpayne@7:         def flush(self) -> bytes:
jpayne@7:             if hasattr(self._obj, "flush"):
jpayne@7:                 return self._obj.flush()  # type: ignore[no-any-return]
jpayne@7:             return b""
jpayne@7: 
jpayne@7: 
jpayne@7: if zstd is not None:
jpayne@7: 
jpayne@7:     class ZstdDecoder(ContentDecoder):
jpayne@7:         def __init__(self) -> None:
jpayne@7:             self._obj = zstd.ZstdDecompressor().decompressobj()
jpayne@7: 
jpayne@7:         def decompress(self, data: bytes) -> bytes:
jpayne@7:             if not data:
jpayne@7:                 return b""
jpayne@7:             data_parts = [self._obj.decompress(data)]
jpayne@7:             while self._obj.eof and self._obj.unused_data:
jpayne@7:                 unused_data = self._obj.unused_data
jpayne@7:                 self._obj = zstd.ZstdDecompressor().decompressobj()
jpayne@7:                 data_parts.append(self._obj.decompress(unused_data))
jpayne@7:             return b"".join(data_parts)
jpayne@7: 
jpayne@7:         def flush(self) -> bytes:
jpayne@7:             ret = self._obj.flush()  # note: this is a no-op
jpayne@7:             if not self._obj.eof:
jpayne@7:                 raise DecodeError("Zstandard data is incomplete")
jpayne@7:             return ret  # type: ignore[no-any-return]
jpayne@7: 
jpayne@7: 
jpayne@7: class MultiDecoder(ContentDecoder):
jpayne@7:     """
jpayne@7:     From RFC7231:
jpayne@7:         If one or more encodings have been applied to a representation, the
jpayne@7:         sender that applied the encodings MUST generate a Content-Encoding
jpayne@7:         header field that lists the content codings in the order in which
jpayne@7:         they were applied.
jpayne@7:     """
jpayne@7: 
jpayne@7:     def __init__(self, modes: str) -> None:
jpayne@7:         self._decoders = [_get_decoder(m.strip()) for m in modes.split(",")]
jpayne@7: 
jpayne@7:     def flush(self) -> bytes:
jpayne@7:         return self._decoders[0].flush()
jpayne@7: 
jpayne@7:     def decompress(self, data: bytes) -> bytes:
jpayne@7:         for d in reversed(self._decoders):
jpayne@7:             data = d.decompress(data)
jpayne@7:         return data
jpayne@7: 
jpayne@7: 
jpayne@7: def _get_decoder(mode: str) -> ContentDecoder:
jpayne@7:     if "," in mode:
jpayne@7:         return MultiDecoder(mode)
jpayne@7: 
jpayne@7:     # According to RFC 9110 section 8.4.1.3, recipients should
jpayne@7:     # consider x-gzip equivalent to gzip
jpayne@7:     if mode in ("gzip", "x-gzip"):
jpayne@7:         return GzipDecoder()
jpayne@7: 
jpayne@7:     if brotli is not None and mode == "br":
jpayne@7:         return BrotliDecoder()
jpayne@7: 
jpayne@7:     if zstd is not None and mode == "zstd":
jpayne@7:         return ZstdDecoder()
jpayne@7: 
jpayne@7:     return DeflateDecoder()
jpayne@7: 
jpayne@7: 
jpayne@7: class BytesQueueBuffer:
jpayne@7:     """Memory-efficient bytes buffer
jpayne@7: 
jpayne@7:     To return decoded data in read() and still follow the BufferedIOBase API, we need a
jpayne@7:     buffer to always return the correct amount of bytes.
jpayne@7: 
jpayne@7:     This buffer should be filled using calls to put()
jpayne@7: 
jpayne@7:     Our maximum memory usage is determined by the sum of the size of:
jpayne@7: 
jpayne@7:      * self.buffer, which contains the full data
jpayne@7:      * the largest chunk that we will copy in get()
jpayne@7: 
jpayne@7:     The worst case scenario is a single chunk, in which case we'll make a full copy of
jpayne@7:     the data inside get().
jpayne@7:     """
jpayne@7: 
jpayne@7:     def __init__(self) -> None:
jpayne@7:         self.buffer: typing.Deque[bytes] = collections.deque()
jpayne@7:         self._size: int = 0
jpayne@7: 
jpayne@7:     def __len__(self) -> int:
jpayne@7:         return self._size
jpayne@7: 
jpayne@7:     def put(self, data: bytes) -> None:
jpayne@7:         self.buffer.append(data)
jpayne@7:         self._size += len(data)
jpayne@7: 
jpayne@7:     def get(self, n: int) -> bytes:
jpayne@7:         if n == 0:
jpayne@7:             return b""
jpayne@7:         elif not self.buffer:
jpayne@7:             raise RuntimeError("buffer is empty")
jpayne@7:         elif n < 0:
jpayne@7:             raise ValueError("n should be > 0")
jpayne@7: 
jpayne@7:         fetched = 0
jpayne@7:         ret = io.BytesIO()
jpayne@7:         while fetched < n:
jpayne@7:             remaining = n - fetched
jpayne@7:             chunk = self.buffer.popleft()
jpayne@7:             chunk_length = len(chunk)
jpayne@7:             if remaining < chunk_length:
jpayne@7:                 left_chunk, right_chunk = chunk[:remaining], chunk[remaining:]
jpayne@7:                 ret.write(left_chunk)
jpayne@7:                 self.buffer.appendleft(right_chunk)
jpayne@7:                 self._size -= remaining
jpayne@7:                 break
jpayne@7:             else:
jpayne@7:                 ret.write(chunk)
jpayne@7:                 self._size -= chunk_length
jpayne@7:             fetched += chunk_length
jpayne@7: 
jpayne@7:             if not self.buffer:
jpayne@7:                 break
jpayne@7: 
jpayne@7:         return ret.getvalue()
jpayne@7: 
jpayne@7:     def get_all(self) -> bytes:
jpayne@7:         buffer = self.buffer
jpayne@7:         if not buffer:
jpayne@7:             assert self._size == 0
jpayne@7:             return b""
jpayne@7:         if len(buffer) == 1:
jpayne@7:             result = buffer.pop()
jpayne@7:         else:
jpayne@7:             ret = io.BytesIO()
jpayne@7:             ret.writelines(buffer.popleft() for _ in range(len(buffer)))
jpayne@7:             result = ret.getvalue()
jpayne@7:         self._size = 0
jpayne@7:         return result
jpayne@7: 
jpayne@7: 
jpayne@7: class BaseHTTPResponse(io.IOBase):
jpayne@7:     CONTENT_DECODERS = ["gzip", "x-gzip", "deflate"]
jpayne@7:     if brotli is not None:
jpayne@7:         CONTENT_DECODERS += ["br"]
jpayne@7:     if zstd is not None:
jpayne@7:         CONTENT_DECODERS += ["zstd"]
jpayne@7:     REDIRECT_STATUSES = [301, 302, 303, 307, 308]
jpayne@7: 
jpayne@7:     DECODER_ERROR_CLASSES: tuple[type[Exception], ...] = (IOError, zlib.error)
jpayne@7:     if brotli is not None:
jpayne@7:         DECODER_ERROR_CLASSES += (brotli.error,)
jpayne@7: 
jpayne@7:     if zstd is not None:
jpayne@7:         DECODER_ERROR_CLASSES += (zstd.ZstdError,)
jpayne@7: 
jpayne@7:     def __init__(
jpayne@7:         self,
jpayne@7:         *,
jpayne@7:         headers: typing.Mapping[str, str] | typing.Mapping[bytes, bytes] | None = None,
jpayne@7:         status: int,
jpayne@7:         version: int,
jpayne@7:         reason: str | None,
jpayne@7:         decode_content: bool,
jpayne@7:         request_url: str | None,
jpayne@7:         retries: Retry | None = None,
jpayne@7:     ) -> None:
jpayne@7:         if isinstance(headers, HTTPHeaderDict):
jpayne@7:             self.headers = headers
jpayne@7:         else:
jpayne@7:             self.headers = HTTPHeaderDict(headers)  # type: ignore[arg-type]
jpayne@7:         self.status = status
jpayne@7:         self.version = version
jpayne@7:         self.reason = reason
jpayne@7:         self.decode_content = decode_content
jpayne@7:         self._has_decoded_content = False
jpayne@7:         self._request_url: str | None = request_url
jpayne@7:         self.retries = retries
jpayne@7: 
jpayne@7:         self.chunked = False
jpayne@7:         tr_enc = self.headers.get("transfer-encoding", "").lower()
jpayne@7:         # Don't incur the penalty of creating a list and then discarding it
jpayne@7:         encodings = (enc.strip() for enc in tr_enc.split(","))
jpayne@7:         if "chunked" in encodings:
jpayne@7:             self.chunked = True
jpayne@7: 
jpayne@7:         self._decoder: ContentDecoder | None = None
jpayne@7:         self.length_remaining: int | None
jpayne@7: 
jpayne@7:     def get_redirect_location(self) -> str | None | Literal[False]:
jpayne@7:         """
jpayne@7:         Should we redirect and where to?
jpayne@7: 
jpayne@7:         :returns: Truthy redirect location string if we got a redirect status
jpayne@7:             code and valid location. ``None`` if redirect status and no
jpayne@7:             location. ``False`` if not a redirect status code.
jpayne@7:         """
jpayne@7:         if self.status in self.REDIRECT_STATUSES:
jpayne@7:             return self.headers.get("location")
jpayne@7:         return False
jpayne@7: 
jpayne@7:     @property
jpayne@7:     def data(self) -> bytes:
jpayne@7:         raise NotImplementedError()
jpayne@7: 
jpayne@7:     def json(self) -> typing.Any:
jpayne@7:         """
jpayne@7:         Parses the body of the HTTP response as JSON.
jpayne@7: 
jpayne@7:         To use a custom JSON decoder pass the result of :attr:`HTTPResponse.data` to the decoder.
jpayne@7: 
jpayne@7:         This method can raise either `UnicodeDecodeError` or `json.JSONDecodeError`.
jpayne@7: 
jpayne@7:         Read more :ref:`here <json>`.
jpayne@7:         """
jpayne@7:         data = self.data.decode("utf-8")
jpayne@7:         return _json.loads(data)
jpayne@7: 
jpayne@7:     @property
jpayne@7:     def url(self) -> str | None:
jpayne@7:         raise NotImplementedError()
jpayne@7: 
jpayne@7:     @url.setter
jpayne@7:     def url(self, url: str | None) -> None:
jpayne@7:         raise NotImplementedError()
jpayne@7: 
jpayne@7:     @property
jpayne@7:     def connection(self) -> BaseHTTPConnection | None:
jpayne@7:         raise NotImplementedError()
jpayne@7: 
jpayne@7:     @property
jpayne@7:     def retries(self) -> Retry | None:
jpayne@7:         return self._retries
jpayne@7: 
jpayne@7:     @retries.setter
jpayne@7:     def retries(self, retries: Retry | None) -> None:
jpayne@7:         # Override the request_url if retries has a redirect location.
jpayne@7:         if retries is not None and retries.history:
jpayne@7:             self.url = retries.history[-1].redirect_location
jpayne@7:         self._retries = retries
jpayne@7: 
jpayne@7:     def stream(
jpayne@7:         self, amt: int | None = 2**16, decode_content: bool | None = None
jpayne@7:     ) -> typing.Iterator[bytes]:
jpayne@7:         raise NotImplementedError()
jpayne@7: 
jpayne@7:     def read(
jpayne@7:         self,
jpayne@7:         amt: int | None = None,
jpayne@7:         decode_content: bool | None = None,
jpayne@7:         cache_content: bool = False,
jpayne@7:     ) -> bytes:
jpayne@7:         raise NotImplementedError()
jpayne@7: 
jpayne@7:     def read1(
jpayne@7:         self,
jpayne@7:         amt: int | None = None,
jpayne@7:         decode_content: bool | None = None,
jpayne@7:     ) -> bytes:
jpayne@7:         raise NotImplementedError()
jpayne@7: 
jpayne@7:     def read_chunked(
jpayne@7:         self,
jpayne@7:         amt: int | None = None,
jpayne@7:         decode_content: bool | None = None,
jpayne@7:     ) -> typing.Iterator[bytes]:
jpayne@7:         raise NotImplementedError()
jpayne@7: 
jpayne@7:     def release_conn(self) -> None:
jpayne@7:         raise NotImplementedError()
jpayne@7: 
jpayne@7:     def drain_conn(self) -> None:
jpayne@7:         raise NotImplementedError()
jpayne@7: 
jpayne@7:     def close(self) -> None:
jpayne@7:         raise NotImplementedError()
jpayne@7: 
jpayne@7:     def _init_decoder(self) -> None:
jpayne@7:         """
jpayne@7:         Set-up the _decoder attribute if necessary.
jpayne@7:         """
jpayne@7:         # Note: content-encoding value should be case-insensitive, per RFC 7230
jpayne@7:         # Section 3.2
jpayne@7:         content_encoding = self.headers.get("content-encoding", "").lower()
jpayne@7:         if self._decoder is None:
jpayne@7:             if content_encoding in self.CONTENT_DECODERS:
jpayne@7:                 self._decoder = _get_decoder(content_encoding)
jpayne@7:             elif "," in content_encoding:
jpayne@7:                 encodings = [
jpayne@7:                     e.strip()
jpayne@7:                     for e in content_encoding.split(",")
jpayne@7:                     if e.strip() in self.CONTENT_DECODERS
jpayne@7:                 ]
jpayne@7:                 if encodings:
jpayne@7:                     self._decoder = _get_decoder(content_encoding)
jpayne@7: 
jpayne@7:     def _decode(
jpayne@7:         self, data: bytes, decode_content: bool | None, flush_decoder: bool
jpayne@7:     ) -> bytes:
jpayne@7:         """
jpayne@7:         Decode the data passed in and potentially flush the decoder.
jpayne@7:         """
jpayne@7:         if not decode_content:
jpayne@7:             if self._has_decoded_content:
jpayne@7:                 raise RuntimeError(
jpayne@7:                     "Calling read(decode_content=False) is not supported after "
jpayne@7:                     "read(decode_content=True) was called."
jpayne@7:                 )
jpayne@7:             return data
jpayne@7: 
jpayne@7:         try:
jpayne@7:             if self._decoder:
jpayne@7:                 data = self._decoder.decompress(data)
jpayne@7:                 self._has_decoded_content = True
jpayne@7:         except self.DECODER_ERROR_CLASSES as e:
jpayne@7:             content_encoding = self.headers.get("content-encoding", "").lower()
jpayne@7:             raise DecodeError(
jpayne@7:                 "Received response with content-encoding: %s, but "
jpayne@7:                 "failed to decode it." % content_encoding,
jpayne@7:                 e,
jpayne@7:             ) from e
jpayne@7:         if flush_decoder:
jpayne@7:             data += self._flush_decoder()
jpayne@7: 
jpayne@7:         return data
jpayne@7: 
jpayne@7:     def _flush_decoder(self) -> bytes:
jpayne@7:         """
jpayne@7:         Flushes the decoder. Should only be called if the decoder is actually
jpayne@7:         being used.
jpayne@7:         """
jpayne@7:         if self._decoder:
jpayne@7:             return self._decoder.decompress(b"") + self._decoder.flush()
jpayne@7:         return b""
jpayne@7: 
jpayne@7:     # Compatibility methods for `io` module
jpayne@7:     def readinto(self, b: bytearray) -> int:
jpayne@7:         temp = self.read(len(b))
jpayne@7:         if len(temp) == 0:
jpayne@7:             return 0
jpayne@7:         else:
jpayne@7:             b[: len(temp)] = temp
jpayne@7:             return len(temp)
jpayne@7: 
jpayne@7:     # Compatibility methods for http.client.HTTPResponse
jpayne@7:     def getheaders(self) -> HTTPHeaderDict:
jpayne@7:         warnings.warn(
jpayne@7:             "HTTPResponse.getheaders() is deprecated and will be removed "
jpayne@7:             "in urllib3 v2.1.0. Instead access HTTPResponse.headers directly.",
jpayne@7:             category=DeprecationWarning,
jpayne@7:             stacklevel=2,
jpayne@7:         )
jpayne@7:         return self.headers
jpayne@7: 
jpayne@7:     def getheader(self, name: str, default: str | None = None) -> str | None:
jpayne@7:         warnings.warn(
jpayne@7:             "HTTPResponse.getheader() is deprecated and will be removed "
jpayne@7:             "in urllib3 v2.1.0. Instead use HTTPResponse.headers.get(name, default).",
jpayne@7:             category=DeprecationWarning,
jpayne@7:             stacklevel=2,
jpayne@7:         )
jpayne@7:         return self.headers.get(name, default)
jpayne@7: 
jpayne@7:     # Compatibility method for http.cookiejar
jpayne@7:     def info(self) -> HTTPHeaderDict:
jpayne@7:         return self.headers
jpayne@7: 
jpayne@7:     def geturl(self) -> str | None:
jpayne@7:         return self.url
jpayne@7: 
jpayne@7: 
jpayne@7: class HTTPResponse(BaseHTTPResponse):
jpayne@7:     """
jpayne@7:     HTTP Response container.
jpayne@7: 
jpayne@7:     Backwards-compatible with :class:`http.client.HTTPResponse` but the response ``body`` is
jpayne@7:     loaded and decoded on-demand when the ``data`` property is accessed.  This
jpayne@7:     class is also compatible with the Python standard library's :mod:`io`
jpayne@7:     module, and can hence be treated as a readable object in the context of that
jpayne@7:     framework.
jpayne@7: 
jpayne@7:     Extra parameters for behaviour not present in :class:`http.client.HTTPResponse`:
jpayne@7: 
jpayne@7:     :param preload_content:
jpayne@7:         If True, the response's body will be preloaded during construction.
jpayne@7: 
jpayne@7:     :param decode_content:
jpayne@7:         If True, will attempt to decode the body based on the
jpayne@7:         'content-encoding' header.
jpayne@7: 
jpayne@7:     :param original_response:
jpayne@7:         When this HTTPResponse wrapper is generated from an :class:`http.client.HTTPResponse`
jpayne@7:         object, it's convenient to include the original for debug purposes. It's
jpayne@7:         otherwise unused.
jpayne@7: 
jpayne@7:     :param retries:
jpayne@7:         The retries contains the last :class:`~urllib3.util.retry.Retry` that
jpayne@7:         was used during the request.
jpayne@7: 
jpayne@7:     :param enforce_content_length:
jpayne@7:         Enforce content length checking. Body returned by server must match
jpayne@7:         value of Content-Length header, if present. Otherwise, raise error.
jpayne@7:     """
jpayne@7: 
jpayne@7:     def __init__(
jpayne@7:         self,
jpayne@7:         body: _TYPE_BODY = "",
jpayne@7:         headers: typing.Mapping[str, str] | typing.Mapping[bytes, bytes] | None = None,
jpayne@7:         status: int = 0,
jpayne@7:         version: int = 0,
jpayne@7:         reason: str | None = None,
jpayne@7:         preload_content: bool = True,
jpayne@7:         decode_content: bool = True,
jpayne@7:         original_response: _HttplibHTTPResponse | None = None,
jpayne@7:         pool: HTTPConnectionPool | None = None,
jpayne@7:         connection: HTTPConnection | None = None,
jpayne@7:         msg: _HttplibHTTPMessage | None = None,
jpayne@7:         retries: Retry | None = None,
jpayne@7:         enforce_content_length: bool = True,
jpayne@7:         request_method: str | None = None,
jpayne@7:         request_url: str | None = None,
jpayne@7:         auto_close: bool = True,
jpayne@7:     ) -> None:
jpayne@7:         super().__init__(
jpayne@7:             headers=headers,
jpayne@7:             status=status,
jpayne@7:             version=version,
jpayne@7:             reason=reason,
jpayne@7:             decode_content=decode_content,
jpayne@7:             request_url=request_url,
jpayne@7:             retries=retries,
jpayne@7:         )
jpayne@7: 
jpayne@7:         self.enforce_content_length = enforce_content_length
jpayne@7:         self.auto_close = auto_close
jpayne@7: 
jpayne@7:         self._body = None
jpayne@7:         self._fp: _HttplibHTTPResponse | None = None
jpayne@7:         self._original_response = original_response
jpayne@7:         self._fp_bytes_read = 0
jpayne@7:         self.msg = msg
jpayne@7: 
jpayne@7:         if body and isinstance(body, (str, bytes)):
jpayne@7:             self._body = body
jpayne@7: 
jpayne@7:         self._pool = pool
jpayne@7:         self._connection = connection
jpayne@7: 
jpayne@7:         if hasattr(body, "read"):
jpayne@7:             self._fp = body  # type: ignore[assignment]
jpayne@7: 
jpayne@7:         # Are we using the chunked-style of transfer encoding?
jpayne@7:         self.chunk_left: int | None = None
jpayne@7: 
jpayne@7:         # Determine length of response
jpayne@7:         self.length_remaining = self._init_length(request_method)
jpayne@7: 
jpayne@7:         # Used to return the correct amount of bytes for partial read()s
jpayne@7:         self._decoded_buffer = BytesQueueBuffer()
jpayne@7: 
jpayne@7:         # If requested, preload the body.
jpayne@7:         if preload_content and not self._body:
jpayne@7:             self._body = self.read(decode_content=decode_content)
jpayne@7: 
jpayne@7:     def release_conn(self) -> None:
jpayne@7:         if not self._pool or not self._connection:
jpayne@7:             return None
jpayne@7: 
jpayne@7:         self._pool._put_conn(self._connection)
jpayne@7:         self._connection = None
jpayne@7: 
jpayne@7:     def drain_conn(self) -> None:
jpayne@7:         """
jpayne@7:         Read and discard any remaining HTTP response data in the response connection.
jpayne@7: 
jpayne@7:         Unread data in the HTTPResponse connection blocks the connection from being released back to the pool.
jpayne@7:         """
jpayne@7:         try:
jpayne@7:             self.read()
jpayne@7:         except (HTTPError, OSError, BaseSSLError, HTTPException):
jpayne@7:             pass
jpayne@7: 
jpayne@7:     @property
jpayne@7:     def data(self) -> bytes:
jpayne@7:         # For backwards-compat with earlier urllib3 0.4 and earlier.
jpayne@7:         if self._body:
jpayne@7:             return self._body  # type: ignore[return-value]
jpayne@7: 
jpayne@7:         if self._fp:
jpayne@7:             return self.read(cache_content=True)
jpayne@7: 
jpayne@7:         return None  # type: ignore[return-value]
jpayne@7: 
jpayne@7:     @property
jpayne@7:     def connection(self) -> HTTPConnection | None:
jpayne@7:         return self._connection
jpayne@7: 
jpayne@7:     def isclosed(self) -> bool:
jpayne@7:         return is_fp_closed(self._fp)
jpayne@7: 
jpayne@7:     def tell(self) -> int:
jpayne@7:         """
jpayne@7:         Obtain the number of bytes pulled over the wire so far. May differ from
jpayne@7:         the amount of content returned by :meth:``urllib3.response.HTTPResponse.read``
jpayne@7:         if bytes are encoded on the wire (e.g, compressed).
jpayne@7:         """
jpayne@7:         return self._fp_bytes_read
jpayne@7: 
jpayne@7:     def _init_length(self, request_method: str | None) -> int | None:
jpayne@7:         """
jpayne@7:         Set initial length value for Response content if available.
jpayne@7:         """
jpayne@7:         length: int | None
jpayne@7:         content_length: str | None = self.headers.get("content-length")
jpayne@7: 
jpayne@7:         if content_length is not None:
jpayne@7:             if self.chunked:
jpayne@7:                 # This Response will fail with an IncompleteRead if it can't be
jpayne@7:                 # received as chunked. This method falls back to attempt reading
jpayne@7:                 # the response before raising an exception.
jpayne@7:                 log.warning(
jpayne@7:                     "Received response with both Content-Length and "
jpayne@7:                     "Transfer-Encoding set. This is expressly forbidden "
jpayne@7:                     "by RFC 7230 sec 3.3.2. Ignoring Content-Length and "
jpayne@7:                     "attempting to process response as Transfer-Encoding: "
jpayne@7:                     "chunked."
jpayne@7:                 )
jpayne@7:                 return None
jpayne@7: 
jpayne@7:             try:
jpayne@7:                 # RFC 7230 section 3.3.2 specifies multiple content lengths can
jpayne@7:                 # be sent in a single Content-Length header
jpayne@7:                 # (e.g. Content-Length: 42, 42). This line ensures the values
jpayne@7:                 # are all valid ints and that as long as the `set` length is 1,
jpayne@7:                 # all values are the same. Otherwise, the header is invalid.
jpayne@7:                 lengths = {int(val) for val in content_length.split(",")}
jpayne@7:                 if len(lengths) > 1:
jpayne@7:                     raise InvalidHeader(
jpayne@7:                         "Content-Length contained multiple "
jpayne@7:                         "unmatching values (%s)" % content_length
jpayne@7:                     )
jpayne@7:                 length = lengths.pop()
jpayne@7:             except ValueError:
jpayne@7:                 length = None
jpayne@7:             else:
jpayne@7:                 if length < 0:
jpayne@7:                     length = None
jpayne@7: 
jpayne@7:         else:  # if content_length is None
jpayne@7:             length = None
jpayne@7: 
jpayne@7:         # Convert status to int for comparison
jpayne@7:         # In some cases, httplib returns a status of "_UNKNOWN"
jpayne@7:         try:
jpayne@7:             status = int(self.status)
jpayne@7:         except ValueError:
jpayne@7:             status = 0
jpayne@7: 
jpayne@7:         # Check for responses that shouldn't include a body
jpayne@7:         if status in (204, 304) or 100 <= status < 200 or request_method == "HEAD":
jpayne@7:             length = 0
jpayne@7: 
jpayne@7:         return length
jpayne@7: 
jpayne@7:     @contextmanager
jpayne@7:     def _error_catcher(self) -> typing.Generator[None, None, None]:
jpayne@7:         """
jpayne@7:         Catch low-level python exceptions, instead re-raising urllib3
jpayne@7:         variants, so that low-level exceptions are not leaked in the
jpayne@7:         high-level api.
jpayne@7: 
jpayne@7:         On exit, release the connection back to the pool.
jpayne@7:         """
jpayne@7:         clean_exit = False
jpayne@7: 
jpayne@7:         try:
jpayne@7:             try:
jpayne@7:                 yield
jpayne@7: 
jpayne@7:             except SocketTimeout as e:
jpayne@7:                 # FIXME: Ideally we'd like to include the url in the ReadTimeoutError but
jpayne@7:                 # there is yet no clean way to get at it from this context.
jpayne@7:                 raise ReadTimeoutError(self._pool, None, "Read timed out.") from e  # type: ignore[arg-type]
jpayne@7: 
jpayne@7:             except BaseSSLError as e:
jpayne@7:                 # FIXME: Is there a better way to differentiate between SSLErrors?
jpayne@7:                 if "read operation timed out" not in str(e):
jpayne@7:                     # SSL errors related to framing/MAC get wrapped and reraised here
jpayne@7:                     raise SSLError(e) from e
jpayne@7: 
jpayne@7:                 raise ReadTimeoutError(self._pool, None, "Read timed out.") from e  # type: ignore[arg-type]
jpayne@7: 
jpayne@7:             except IncompleteRead as e:
jpayne@7:                 if (
jpayne@7:                     e.expected is not None
jpayne@7:                     and e.partial is not None
jpayne@7:                     and e.expected == -e.partial
jpayne@7:                 ):
jpayne@7:                     arg = "Response may not contain content."
jpayne@7:                 else:
jpayne@7:                     arg = f"Connection broken: {e!r}"
jpayne@7:                 raise ProtocolError(arg, e) from e
jpayne@7: 
jpayne@7:             except (HTTPException, OSError) as e:
jpayne@7:                 raise ProtocolError(f"Connection broken: {e!r}", e) from e
jpayne@7: 
jpayne@7:             # If no exception is thrown, we should avoid cleaning up
jpayne@7:             # unnecessarily.
jpayne@7:             clean_exit = True
jpayne@7:         finally:
jpayne@7:             # If we didn't terminate cleanly, we need to throw away our
jpayne@7:             # connection.
jpayne@7:             if not clean_exit:
jpayne@7:                 # The response may not be closed but we're not going to use it
jpayne@7:                 # anymore so close it now to ensure that the connection is
jpayne@7:                 # released back to the pool.
jpayne@7:                 if self._original_response:
jpayne@7:                     self._original_response.close()
jpayne@7: 
jpayne@7:                 # Closing the response may not actually be sufficient to close
jpayne@7:                 # everything, so if we have a hold of the connection close that
jpayne@7:                 # too.
jpayne@7:                 if self._connection:
jpayne@7:                     self._connection.close()
jpayne@7: 
jpayne@7:             # If we hold the original response but it's closed now, we should
jpayne@7:             # return the connection back to the pool.
jpayne@7:             if self._original_response and self._original_response.isclosed():
jpayne@7:                 self.release_conn()
jpayne@7: 
jpayne@7:     def _fp_read(
jpayne@7:         self,
jpayne@7:         amt: int | None = None,
jpayne@7:         *,
jpayne@7:         read1: bool = False,
jpayne@7:     ) -> bytes:
jpayne@7:         """
jpayne@7:         Read a response with the thought that reading the number of bytes
jpayne@7:         larger than can fit in a 32-bit int at a time via SSL in some
jpayne@7:         known cases leads to an overflow error that has to be prevented
jpayne@7:         if `amt` or `self.length_remaining` indicate that a problem may
jpayne@7:         happen.
jpayne@7: 
jpayne@7:         The known cases:
jpayne@7:           * 3.8 <= CPython < 3.9.7 because of a bug
jpayne@7:             https://github.com/urllib3/urllib3/issues/2513#issuecomment-1152559900.
jpayne@7:           * urllib3 injected with pyOpenSSL-backed SSL-support.
jpayne@7:           * CPython < 3.10 only when `amt` does not fit 32-bit int.
jpayne@7:         """
jpayne@7:         assert self._fp
jpayne@7:         c_int_max = 2**31 - 1
jpayne@7:         if (
jpayne@7:             (amt and amt > c_int_max)
jpayne@7:             or (
jpayne@7:                 amt is None
jpayne@7:                 and self.length_remaining
jpayne@7:                 and self.length_remaining > c_int_max
jpayne@7:             )
jpayne@7:         ) and (util.IS_PYOPENSSL or sys.version_info < (3, 10)):
jpayne@7:             if read1:
jpayne@7:                 return self._fp.read1(c_int_max)
jpayne@7:             buffer = io.BytesIO()
jpayne@7:             # Besides `max_chunk_amt` being a maximum chunk size, it
jpayne@7:             # affects memory overhead of reading a response by this
jpayne@7:             # method in CPython.
jpayne@7:             # `c_int_max` equal to 2 GiB - 1 byte is the actual maximum
jpayne@7:             # chunk size that does not lead to an overflow error, but
jpayne@7:             # 256 MiB is a compromise.
jpayne@7:             max_chunk_amt = 2**28
jpayne@7:             while amt is None or amt != 0:
jpayne@7:                 if amt is not None:
jpayne@7:                     chunk_amt = min(amt, max_chunk_amt)
jpayne@7:                     amt -= chunk_amt
jpayne@7:                 else:
jpayne@7:                     chunk_amt = max_chunk_amt
jpayne@7:                 data = self._fp.read(chunk_amt)
jpayne@7:                 if not data:
jpayne@7:                     break
jpayne@7:                 buffer.write(data)
jpayne@7:                 del data  # to reduce peak memory usage by `max_chunk_amt`.
jpayne@7:             return buffer.getvalue()
jpayne@7:         elif read1:
jpayne@7:             return self._fp.read1(amt) if amt is not None else self._fp.read1()
jpayne@7:         else:
jpayne@7:             # StringIO doesn't like amt=None
jpayne@7:             return self._fp.read(amt) if amt is not None else self._fp.read()
jpayne@7: 
jpayne@7:     def _raw_read(
jpayne@7:         self,
jpayne@7:         amt: int | None = None,
jpayne@7:         *,
jpayne@7:         read1: bool = False,
jpayne@7:     ) -> bytes:
jpayne@7:         """
jpayne@7:         Reads `amt` of bytes from the socket.
jpayne@7:         """
jpayne@7:         if self._fp is None:
jpayne@7:             return None  # type: ignore[return-value]
jpayne@7: 
jpayne@7:         fp_closed = getattr(self._fp, "closed", False)
jpayne@7: 
jpayne@7:         with self._error_catcher():
jpayne@7:             data = self._fp_read(amt, read1=read1) if not fp_closed else b""
jpayne@7:             if amt is not None and amt != 0 and not data:
jpayne@7:                 # Platform-specific: Buggy versions of Python.
jpayne@7:                 # Close the connection when no data is returned
jpayne@7:                 #
jpayne@7:                 # This is redundant to what httplib/http.client _should_
jpayne@7:                 # already do.  However, versions of python released before
jpayne@7:                 # December 15, 2012 (http://bugs.python.org/issue16298) do
jpayne@7:                 # not properly close the connection in all cases. There is
jpayne@7:                 # no harm in redundantly calling close.
jpayne@7:                 self._fp.close()
jpayne@7:                 if (
jpayne@7:                     self.enforce_content_length
jpayne@7:                     and self.length_remaining is not None
jpayne@7:                     and self.length_remaining != 0
jpayne@7:                 ):
jpayne@7:                     # This is an edge case that httplib failed to cover due
jpayne@7:                     # to concerns of backward compatibility. We're
jpayne@7:                     # addressing it here to make sure IncompleteRead is
jpayne@7:                     # raised during streaming, so all calls with incorrect
jpayne@7:                     # Content-Length are caught.
jpayne@7:                     raise IncompleteRead(self._fp_bytes_read, self.length_remaining)
jpayne@7:             elif read1 and (
jpayne@7:                 (amt != 0 and not data) or self.length_remaining == len(data)
jpayne@7:             ):
jpayne@7:                 # All data has been read, but `self._fp.read1` in
jpayne@7:                 # CPython 3.12 and older doesn't always close
jpayne@7:                 # `http.client.HTTPResponse`, so we close it here.
jpayne@7:                 # See https://github.com/python/cpython/issues/113199
jpayne@7:                 self._fp.close()
jpayne@7: 
jpayne@7:         if data:
jpayne@7:             self._fp_bytes_read += len(data)
jpayne@7:             if self.length_remaining is not None:
jpayne@7:                 self.length_remaining -= len(data)
jpayne@7:         return data
jpayne@7: 
jpayne@7:     def read(
jpayne@7:         self,
jpayne@7:         amt: int | None = None,
jpayne@7:         decode_content: bool | None = None,
jpayne@7:         cache_content: bool = False,
jpayne@7:     ) -> bytes:
jpayne@7:         """
jpayne@7:         Similar to :meth:`http.client.HTTPResponse.read`, but with two additional
jpayne@7:         parameters: ``decode_content`` and ``cache_content``.
jpayne@7: 
jpayne@7:         :param amt:
jpayne@7:             How much of the content to read. If specified, caching is skipped
jpayne@7:             because it doesn't make sense to cache partial content as the full
jpayne@7:             response.
jpayne@7: 
jpayne@7:         :param decode_content:
jpayne@7:             If True, will attempt to decode the body based on the
jpayne@7:             'content-encoding' header.
jpayne@7: 
jpayne@7:         :param cache_content:
jpayne@7:             If True, will save the returned data such that the same result is
jpayne@7:             returned despite of the state of the underlying file object. This
jpayne@7:             is useful if you want the ``.data`` property to continue working
jpayne@7:             after having ``.read()`` the file object. (Overridden if ``amt`` is
jpayne@7:             set.)
jpayne@7:         """
jpayne@7:         self._init_decoder()
jpayne@7:         if decode_content is None:
jpayne@7:             decode_content = self.decode_content
jpayne@7: 
jpayne@7:         if amt is not None:
jpayne@7:             cache_content = False
jpayne@7: 
jpayne@7:             if len(self._decoded_buffer) >= amt:
jpayne@7:                 return self._decoded_buffer.get(amt)
jpayne@7: 
jpayne@7:         data = self._raw_read(amt)
jpayne@7: 
jpayne@7:         flush_decoder = amt is None or (amt != 0 and not data)
jpayne@7: 
jpayne@7:         if not data and len(self._decoded_buffer) == 0:
jpayne@7:             return data
jpayne@7: 
jpayne@7:         if amt is None:
jpayne@7:             data = self._decode(data, decode_content, flush_decoder)
jpayne@7:             if cache_content:
jpayne@7:                 self._body = data
jpayne@7:         else:
jpayne@7:             # do not waste memory on buffer when not decoding
jpayne@7:             if not decode_content:
jpayne@7:                 if self._has_decoded_content:
jpayne@7:                     raise RuntimeError(
jpayne@7:                         "Calling read(decode_content=False) is not supported after "
jpayne@7:                         "read(decode_content=True) was called."
jpayne@7:                     )
jpayne@7:                 return data
jpayne@7: 
jpayne@7:             decoded_data = self._decode(data, decode_content, flush_decoder)
jpayne@7:             self._decoded_buffer.put(decoded_data)
jpayne@7: 
jpayne@7:             while len(self._decoded_buffer) < amt and data:
jpayne@7:                 # TODO make sure to initially read enough data to get past the headers
jpayne@7:                 # For example, the GZ file header takes 10 bytes, we don't want to read
jpayne@7:                 # it one byte at a time
jpayne@7:                 data = self._raw_read(amt)
jpayne@7:                 decoded_data = self._decode(data, decode_content, flush_decoder)
jpayne@7:                 self._decoded_buffer.put(decoded_data)
jpayne@7:             data = self._decoded_buffer.get(amt)
jpayne@7: 
jpayne@7:         return data
jpayne@7: 
jpayne@7:     def read1(
jpayne@7:         self,
jpayne@7:         amt: int | None = None,
jpayne@7:         decode_content: bool | None = None,
jpayne@7:     ) -> bytes:
jpayne@7:         """
jpayne@7:         Similar to ``http.client.HTTPResponse.read1`` and documented
jpayne@7:         in :meth:`io.BufferedReader.read1`, but with an additional parameter:
jpayne@7:         ``decode_content``.
jpayne@7: 
jpayne@7:         :param amt:
jpayne@7:             How much of the content to read.
jpayne@7: 
jpayne@7:         :param decode_content:
jpayne@7:             If True, will attempt to decode the body based on the
jpayne@7:             'content-encoding' header.
jpayne@7:         """
jpayne@7:         if decode_content is None:
jpayne@7:             decode_content = self.decode_content
jpayne@7:         # try and respond without going to the network
jpayne@7:         if self._has_decoded_content:
jpayne@7:             if not decode_content:
jpayne@7:                 raise RuntimeError(
jpayne@7:                     "Calling read1(decode_content=False) is not supported after "
jpayne@7:                     "read1(decode_content=True) was called."
jpayne@7:                 )
jpayne@7:             if len(self._decoded_buffer) > 0:
jpayne@7:                 if amt is None:
jpayne@7:                     return self._decoded_buffer.get_all()
jpayne@7:                 return self._decoded_buffer.get(amt)
jpayne@7:         if amt == 0:
jpayne@7:             return b""
jpayne@7: 
jpayne@7:         # FIXME, this method's type doesn't say returning None is possible
jpayne@7:         data = self._raw_read(amt, read1=True)
jpayne@7:         if not decode_content or data is None:
jpayne@7:             return data
jpayne@7: 
jpayne@7:         self._init_decoder()
jpayne@7:         while True:
jpayne@7:             flush_decoder = not data
jpayne@7:             decoded_data = self._decode(data, decode_content, flush_decoder)
jpayne@7:             self._decoded_buffer.put(decoded_data)
jpayne@7:             if decoded_data or flush_decoder:
jpayne@7:                 break
jpayne@7:             data = self._raw_read(8192, read1=True)
jpayne@7: 
jpayne@7:         if amt is None:
jpayne@7:             return self._decoded_buffer.get_all()
jpayne@7:         return self._decoded_buffer.get(amt)
jpayne@7: 
jpayne@7:     def stream(
jpayne@7:         self, amt: int | None = 2**16, decode_content: bool | None = None
jpayne@7:     ) -> typing.Generator[bytes, None, None]:
jpayne@7:         """
jpayne@7:         A generator wrapper for the read() method. A call will block until
jpayne@7:         ``amt`` bytes have been read from the connection or until the
jpayne@7:         connection is closed.
jpayne@7: 
jpayne@7:         :param amt:
jpayne@7:             How much of the content to read. The generator will return up to
jpayne@7:             much data per iteration, but may return less. This is particularly
jpayne@7:             likely when using compressed data. However, the empty string will
jpayne@7:             never be returned.
jpayne@7: 
jpayne@7:         :param decode_content:
jpayne@7:             If True, will attempt to decode the body based on the
jpayne@7:             'content-encoding' header.
jpayne@7:         """
jpayne@7:         if self.chunked and self.supports_chunked_reads():
jpayne@7:             yield from self.read_chunked(amt, decode_content=decode_content)
jpayne@7:         else:
jpayne@7:             while not is_fp_closed(self._fp) or len(self._decoded_buffer) > 0:
jpayne@7:                 data = self.read(amt=amt, decode_content=decode_content)
jpayne@7: 
jpayne@7:                 if data:
jpayne@7:                     yield data
jpayne@7: 
jpayne@7:     # Overrides from io.IOBase
jpayne@7:     def readable(self) -> bool:
jpayne@7:         return True
jpayne@7: 
jpayne@7:     def close(self) -> None:
jpayne@7:         if not self.closed and self._fp:
jpayne@7:             self._fp.close()
jpayne@7: 
jpayne@7:         if self._connection:
jpayne@7:             self._connection.close()
jpayne@7: 
jpayne@7:         if not self.auto_close:
jpayne@7:             io.IOBase.close(self)
jpayne@7: 
jpayne@7:     @property
jpayne@7:     def closed(self) -> bool:
jpayne@7:         if not self.auto_close:
jpayne@7:             return io.IOBase.closed.__get__(self)  # type: ignore[no-any-return]
jpayne@7:         elif self._fp is None:
jpayne@7:             return True
jpayne@7:         elif hasattr(self._fp, "isclosed"):
jpayne@7:             return self._fp.isclosed()
jpayne@7:         elif hasattr(self._fp, "closed"):
jpayne@7:             return self._fp.closed
jpayne@7:         else:
jpayne@7:             return True
jpayne@7: 
jpayne@7:     def fileno(self) -> int:
jpayne@7:         if self._fp is None:
jpayne@7:             raise OSError("HTTPResponse has no file to get a fileno from")
jpayne@7:         elif hasattr(self._fp, "fileno"):
jpayne@7:             return self._fp.fileno()
jpayne@7:         else:
jpayne@7:             raise OSError(
jpayne@7:                 "The file-like object this HTTPResponse is wrapped "
jpayne@7:                 "around has no file descriptor"
jpayne@7:             )
jpayne@7: 
jpayne@7:     def flush(self) -> None:
jpayne@7:         if (
jpayne@7:             self._fp is not None
jpayne@7:             and hasattr(self._fp, "flush")
jpayne@7:             and not getattr(self._fp, "closed", False)
jpayne@7:         ):
jpayne@7:             return self._fp.flush()
jpayne@7: 
jpayne@7:     def supports_chunked_reads(self) -> bool:
jpayne@7:         """
jpayne@7:         Checks if the underlying file-like object looks like a
jpayne@7:         :class:`http.client.HTTPResponse` object. We do this by testing for
jpayne@7:         the fp attribute. If it is present we assume it returns raw chunks as
jpayne@7:         processed by read_chunked().
jpayne@7:         """
jpayne@7:         return hasattr(self._fp, "fp")
jpayne@7: 
jpayne@7:     def _update_chunk_length(self) -> None:
jpayne@7:         # First, we'll figure out length of a chunk and then
jpayne@7:         # we'll try to read it from socket.
jpayne@7:         if self.chunk_left is not None:
jpayne@7:             return None
jpayne@7:         line = self._fp.fp.readline()  # type: ignore[union-attr]
jpayne@7:         line = line.split(b";", 1)[0]
jpayne@7:         try:
jpayne@7:             self.chunk_left = int(line, 16)
jpayne@7:         except ValueError:
jpayne@7:             self.close()
jpayne@7:             if line:
jpayne@7:                 # Invalid chunked protocol response, abort.
jpayne@7:                 raise InvalidChunkLength(self, line) from None
jpayne@7:             else:
jpayne@7:                 # Truncated at start of next chunk
jpayne@7:                 raise ProtocolError("Response ended prematurely") from None
jpayne@7: 
jpayne@7:     def _handle_chunk(self, amt: int | None) -> bytes:
jpayne@7:         returned_chunk = None
jpayne@7:         if amt is None:
jpayne@7:             chunk = self._fp._safe_read(self.chunk_left)  # type: ignore[union-attr]
jpayne@7:             returned_chunk = chunk
jpayne@7:             self._fp._safe_read(2)  # type: ignore[union-attr] # Toss the CRLF at the end of the chunk.
jpayne@7:             self.chunk_left = None
jpayne@7:         elif self.chunk_left is not None and amt < self.chunk_left:
jpayne@7:             value = self._fp._safe_read(amt)  # type: ignore[union-attr]
jpayne@7:             self.chunk_left = self.chunk_left - amt
jpayne@7:             returned_chunk = value
jpayne@7:         elif amt == self.chunk_left:
jpayne@7:             value = self._fp._safe_read(amt)  # type: ignore[union-attr]
jpayne@7:             self._fp._safe_read(2)  # type: ignore[union-attr] # Toss the CRLF at the end of the chunk.
jpayne@7:             self.chunk_left = None
jpayne@7:             returned_chunk = value
jpayne@7:         else:  # amt > self.chunk_left
jpayne@7:             returned_chunk = self._fp._safe_read(self.chunk_left)  # type: ignore[union-attr]
jpayne@7:             self._fp._safe_read(2)  # type: ignore[union-attr] # Toss the CRLF at the end of the chunk.
jpayne@7:             self.chunk_left = None
jpayne@7:         return returned_chunk  # type: ignore[no-any-return]
jpayne@7: 
jpayne@7:     def read_chunked(
jpayne@7:         self, amt: int | None = None, decode_content: bool | None = None
jpayne@7:     ) -> typing.Generator[bytes, None, None]:
jpayne@7:         """
jpayne@7:         Similar to :meth:`HTTPResponse.read`, but with an additional
jpayne@7:         parameter: ``decode_content``.
jpayne@7: 
jpayne@7:         :param amt:
jpayne@7:             How much of the content to read. If specified, caching is skipped
jpayne@7:             because it doesn't make sense to cache partial content as the full
jpayne@7:             response.
jpayne@7: 
jpayne@7:         :param decode_content:
jpayne@7:             If True, will attempt to decode the body based on the
jpayne@7:             'content-encoding' header.
jpayne@7:         """
jpayne@7:         self._init_decoder()
jpayne@7:         # FIXME: Rewrite this method and make it a class with a better structured logic.
jpayne@7:         if not self.chunked:
jpayne@7:             raise ResponseNotChunked(
jpayne@7:                 "Response is not chunked. "
jpayne@7:                 "Header 'transfer-encoding: chunked' is missing."
jpayne@7:             )
jpayne@7:         if not self.supports_chunked_reads():
jpayne@7:             raise BodyNotHttplibCompatible(
jpayne@7:                 "Body should be http.client.HTTPResponse like. "
jpayne@7:                 "It should have have an fp attribute which returns raw chunks."
jpayne@7:             )
jpayne@7: 
jpayne@7:         with self._error_catcher():
jpayne@7:             # Don't bother reading the body of a HEAD request.
jpayne@7:             if self._original_response and is_response_to_head(self._original_response):
jpayne@7:                 self._original_response.close()
jpayne@7:                 return None
jpayne@7: 
jpayne@7:             # If a response is already read and closed
jpayne@7:             # then return immediately.
jpayne@7:             if self._fp.fp is None:  # type: ignore[union-attr]
jpayne@7:                 return None
jpayne@7: 
jpayne@7:             while True:
jpayne@7:                 self._update_chunk_length()
jpayne@7:                 if self.chunk_left == 0:
jpayne@7:                     break
jpayne@7:                 chunk = self._handle_chunk(amt)
jpayne@7:                 decoded = self._decode(
jpayne@7:                     chunk, decode_content=decode_content, flush_decoder=False
jpayne@7:                 )
jpayne@7:                 if decoded:
jpayne@7:                     yield decoded
jpayne@7: 
jpayne@7:             if decode_content:
jpayne@7:                 # On CPython and PyPy, we should never need to flush the
jpayne@7:                 # decoder. However, on Jython we *might* need to, so
jpayne@7:                 # lets defensively do it anyway.
jpayne@7:                 decoded = self._flush_decoder()
jpayne@7:                 if decoded:  # Platform-specific: Jython.
jpayne@7:                     yield decoded
jpayne@7: 
jpayne@7:             # Chunk content ends with \r\n: discard it.
jpayne@7:             while self._fp is not None:
jpayne@7:                 line = self._fp.fp.readline()
jpayne@7:                 if not line:
jpayne@7:                     # Some sites may not end with '\r\n'.
jpayne@7:                     break
jpayne@7:                 if line == b"\r\n":
jpayne@7:                     break
jpayne@7: 
jpayne@7:             # We read everything; close the "file".
jpayne@7:             if self._original_response:
jpayne@7:                 self._original_response.close()
jpayne@7: 
jpayne@7:     @property
jpayne@7:     def url(self) -> str | None:
jpayne@7:         """
jpayne@7:         Returns the URL that was the source of this response.
jpayne@7:         If the request that generated this response redirected, this method
jpayne@7:         will return the final redirect location.
jpayne@7:         """
jpayne@7:         return self._request_url
jpayne@7: 
jpayne@7:     @url.setter
jpayne@7:     def url(self, url: str) -> None:
jpayne@7:         self._request_url = url
jpayne@7: 
jpayne@7:     def __iter__(self) -> typing.Iterator[bytes]:
jpayne@7:         buffer: list[bytes] = []
jpayne@7:         for chunk in self.stream(decode_content=True):
jpayne@7:             if b"\n" in chunk:
jpayne@7:                 chunks = chunk.split(b"\n")
jpayne@7:                 yield b"".join(buffer) + chunks[0] + b"\n"
jpayne@7:                 for x in chunks[1:-1]:
jpayne@7:                     yield x + b"\n"
jpayne@7:                 if chunks[-1]:
jpayne@7:                     buffer = [chunks[-1]]
jpayne@7:                 else:
jpayne@7:                     buffer = []
jpayne@7:             else:
jpayne@7:                 buffer.append(chunk)
jpayne@7:         if buffer:
jpayne@7:             yield b"".join(buffer)