Class BrotliDecoder

src/urllib3/response.py:197–237

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)

        # Requires Brotli >= 1.2.0 for `output_buffer_limit`.
        def _decompress(self, data: bytes, output_buffer_limit: int = -1) -> bytes:
            raise NotImplementedError()

        def decompress(self, data: bytes, max_length: int = -1) -> bytes:
            try:
                if max_length > 0:
                    return self._decompress(data, output_buffer_limit=max_length)
                else:
                    return self._decompress(data)
            except TypeError:
                # Fallback for Brotli/brotlicffi/brotlipy versions without
                # the `output_buffer_limit` parameter.
                warnings.warn(
                    "Brotli >= 1.2.0 is required to prevent decompression bombs.",
                    DependencyWarning,
                )
                return self._decompress(data)

        @property
        def has_unconsumed_tail(self) -> bool:
            try:
                return not self._obj.can_accept_more_data()
            except AttributeError:
                return False

        def flush(self) -> bytes:
            if hasattr(self._obj, "flush"):
                return self._obj.flush()  # type: ignore[no-any-return]
            return b""


try: