pgzip

import pgzip import os import tempfile # Create some dummy data original_data = b"This is a test string that will be compressed using pgzip. " * 1000 with tempfile.TemporaryDirectory() as tmpdir: filepath_gz = os.path.join(tmpdir, "test_data.txt.gz") # 1. Compress data to a file using 4 threads and 1MB blocks print(f"Compressing data to {filepath_gz}...") with pgzip.open(filepath_gz, "wb", thread=4, blocksize=2**20) as f_out: f_out.write(original_data) print(f"Compressed file size: {os.path.getsize(filepath_gz)} bytes") # 2. Decompress data from the file using 4 threads print(f"Decompressing data from {filepath_gz}...") with pgzip.open(filepath_gz, "rb", thread=4) as f_in: decompressed_data_file = f_in.read() assert original_data == decompressed_data_file print("File compression and decompression successful!") # 3. In-memory compression and decompression using default threads print("\nPerforming in-memory compression/decompression...") compressed_bytes = pgzip.compress(original_data, compresslevel=6) decompressed_bytes = pgzip.decompress(compressed_bytes) assert original_data == decompressed_bytes print(f"In-memory compression/decompression successful! Compressed size: {len(compressed_bytes)} bytes")