(data: bytes, *, max_size: int = 0)
| 77 | |
| 78 | |
| 79 | def _unzstd(data: bytes, *, max_size: int = 0) -> bytes: |
| 80 | decompressor = zstandard.ZstdDecompressor() |
| 81 | stream_reader = decompressor.stream_reader(BytesIO(data)) |
| 82 | output_stream = BytesIO() |
| 83 | output_chunk = b"." |
| 84 | decompressed_size = 0 |
| 85 | while output_chunk: |
| 86 | output_chunk = stream_reader.read(_CHUNK_SIZE) |
| 87 | decompressed_size += len(output_chunk) |
| 88 | _check_max_size(decompressed_size, max_size) |
| 89 | output_stream.write(output_chunk) |
| 90 | return output_stream.getvalue() |
no test coverage detected