(self, data: bytes, max_length: int = -1)
| 71 | self._obj = zlib.decompressobj() |
| 72 | |
| 73 | def decompress(self, data: bytes, max_length: int = -1) -> bytes: |
| 74 | data = self._unfed_data + data |
| 75 | self._unfed_data = b"" |
| 76 | if not data and not self._obj.unconsumed_tail: |
| 77 | return data |
| 78 | original_max_length = max_length |
| 79 | if original_max_length < 0: |
| 80 | max_length = 0 |
| 81 | elif original_max_length == 0: |
| 82 | # We should not pass 0 to the zlib decompressor because 0 is |
| 83 | # the default value that will make zlib decompress without a |
| 84 | # length limit. |
| 85 | # Data should be stored for subsequent calls. |
| 86 | self._unfed_data = data |
| 87 | return b"" |
| 88 | |
| 89 | # Subsequent calls always reuse `self._obj`. zlib requires |
| 90 | # passing the unconsumed tail if decompression is to continue. |
| 91 | if not self._first_try: |
| 92 | return self._obj.decompress( |
| 93 | self._obj.unconsumed_tail + data, max_length=max_length |
| 94 | ) |
| 95 | |
| 96 | # First call tries with RFC 1950 ZLIB format. |
| 97 | self._first_try_data += data |
| 98 | try: |
| 99 | decompressed = self._obj.decompress(data, max_length=max_length) |
| 100 | if decompressed: |
| 101 | self._first_try = False |
| 102 | self._first_try_data = b"" |
| 103 | return decompressed |
| 104 | # On failure, it falls back to RFC 1951 DEFLATE format. |
| 105 | except zlib.error: |
| 106 | self._first_try = False |
| 107 | self._obj = zlib.decompressobj(-zlib.MAX_WBITS) |
| 108 | try: |
| 109 | return self.decompress( |
| 110 | self._first_try_data, max_length=original_max_length |
| 111 | ) |
| 112 | finally: |
| 113 | self._first_try_data = b"" |
| 114 | |
| 115 | @property |
| 116 | def has_unconsumed_tail(self) -> bool: |
nothing calls this directly
no test coverage detected