|
class _GzipDecoder(urllib3.response.GzipDecoder): |
|
"""Custom subclass of ``urllib3`` decoder for ``gzip``-ed bytes. |
|
|
|
Allows a checksum function to see the compressed bytes before they are |
|
decoded. This way the checksum of the compressed value can be computed. |
|
|
|
Args: |
|
checksum (object): |
|
A checksum which will be updated with compressed bytes. |
|
""" |
|
|
|
def __init__(self, checksum): |
|
super().__init__() |
|
self._checksum = checksum |
|
|
|
def decompress(self, data): |
|
"""Decompress the bytes. |
|
|
|
Args: |
|
data (bytes): The compressed bytes to be decompressed. |
|
|
|
Returns: |
|
bytes: The decompressed bytes from ``data``. |
|
""" |
|
self._checksum.update(data) |
|
return super().decompress(data) |
|
|
|
|
|
# urllib3.response.BrotliDecoder might not exist depending on whether brotli is |
|
# installed. |
|
if hasattr(urllib3.response, "BrotliDecoder"): |
|
|
|
class _BrotliDecoder: |
|
"""Handler for ``brotli`` encoded bytes. |
|
|
|
Allows a checksum function to see the compressed bytes before they are |
|
decoded. This way the checksum of the compressed value can be computed. |
|
|
|
Because BrotliDecoder's decompress method is dynamically created in |
|
urllib3, a subclass is not practical. Instead, this class creates a |
|
captive urllib3.requests.BrotliDecoder instance and acts as a proxy. |
|
|
|
Args: |
|
checksum (object): |
|
A checksum which will be updated with compressed bytes. |
|
""" |
|
|
|
def __init__(self, checksum): |
|
self._decoder = urllib3.response.BrotliDecoder() |
|
self._checksum = checksum |
|
|
|
def decompress(self, data): |
|
"""Decompress the bytes. |
|
|
|
Args: |
|
data (bytes): The compressed bytes to be decompressed. |
|
|
|
Returns: |
|
bytes: The decompressed bytes from ``data``. |
|
""" |
|
self._checksum.update(data) |
|
return self._decoder.decompress(data) |
|
|
|
def flush(self): |
|
return self._decoder.flush() |
|
|
|
else: # pragma: NO COVER |
|
_BrotliDecoder = None # type: ignore # pragma: NO COVER |
