Decompress a gzip compressed string in one shot. Return the decompressed string.
(data)
| 639 | |
| 640 | |
| 641 | def decompress(data): |
| 642 | """Decompress a gzip compressed string in one shot. |
| 643 | Return the decompressed string. |
| 644 | """ |
| 645 | decompressed_members = [] |
| 646 | while True: |
| 647 | fp = io.BytesIO(data) |
| 648 | if _read_gzip_header(fp) is None: |
| 649 | return b"".join(decompressed_members) |
| 650 | # Use a zlib raw deflate compressor |
| 651 | do = zlib.decompressobj(wbits=-zlib.MAX_WBITS) |
| 652 | # Read all the data except the header |
| 653 | decompressed = do.decompress(data[fp.tell():]) |
| 654 | if not do.eof or len(do.unused_data) < 8: |
| 655 | raise EOFError("Compressed file ended before the end-of-stream " |
| 656 | "marker was reached") |
| 657 | crc, length = struct.unpack("<II", do.unused_data[:8]) |
| 658 | if crc != zlib.crc32(decompressed): |
| 659 | raise BadGzipFile("CRC check failed") |
| 660 | if length != (len(decompressed) & 0xffffffff): |
| 661 | raise BadGzipFile("Incorrect length of data produced") |
| 662 | decompressed_members.append(decompressed) |
| 663 | data = do.unused_data[8:].lstrip(b"\x00") |
| 664 | |
| 665 | |
| 666 | def main(): |
nothing calls this directly
no test coverage detected