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