Install & Compatibility
Where this runs
tested against v1.2.0 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
muslpy 3.10–3.910 runs
installs and imports cleanly · install 0.0s · import 0.209s · 19.2MB
glibcpy 3.10–3.910 runs
installs and imports cleanly · install 2.5s · import 0.184s · 20MB
17MB installed
● package 17MB
Code
Verified usage
Verified import paths — ran on the pinned version, not inferred.
GZIPCompressedStream
✓ from gzip_stream import GZIPCompressedStream
GZIPDecompressedStream
✓ from gzip_stream import GZIPDecompressedStream
AsyncGZIPCompressedStream
✓ from gzip_stream import AsyncGZIPCompressedStream
For asynchronous compression in asyncio applications.
AsyncGZIPDecompressedStream
✓ from gzip_stream import AsyncGZIPDecompressedStream
For asynchronous decompression in asyncio applications.
This quickstart demonstrates synchronous compression and decompression of a byte stream using `gzip-stream`. It shows how to create an iterable of bytes, pass it to `GZIPCompressedStream`, and then take the resulting compressed chunks and pass them to `GZIPDecompressedStream` to retrieve the original data.
import gzip_stream
import io
# Original data (must be bytes)
original_data = b"This is some data that will be compressed and then decompressed using gzip-stream." * 5 # Make it longer
# --- Compression Example (Synchronous) ---
# Create an iterable of bytes. In real apps, this could be reading from a file or network.
def generate_bytes_for_compression(data_bytes):
chunk_size = 15 # Arbitrary chunk size
for i in range(0, len(data_bytes), chunk_size):
yield data_bytes[i : i + chunk_size]
# Compress the data by passing the iterable to GZIPCompressedStream
compressed_chunks = list(gzip_stream.GZIPCompressedStream(generate_bytes_for_compression(original_data)))
compressed_data = b"".join(compressed_chunks)
print(f"Original size: {len(original_data)} bytes")
print(f"Compressed size: {len(compressed_data)} bytes")
# --- Decompression Example (Synchronous) ---
# Create an iterable of compressed bytes.
def generate_bytes_for_decompression(compressed_data_bytes):
chunk_size = 20 # Arbitrary chunk size for decompression
for i in range(0, len(compressed_data_bytes), chunk_size):
yield compressed_data_bytes[i : i + chunk_size]
# Decompress the data by passing the iterable to GZIPDecompressedStream
decompressed_chunks = list(gzip_stream.GZIPDecompressedStream(generate_bytes_for_decompression(compressed_data)))
decompressed_data = b"".join(decompressed_chunks)
print(f"Decompressed data (first 50 bytes): {decompressed_data[:50].decode()}")
# Verify data integrity
assert original_data == decompressed_data
print("\nSuccess: Original and decompressed data match!")
Errors
Common errors & fixes
TypeError: a bytes-like object is required, not 'str'
Attempting to pass string data directly to `GZIPCompressedStream` or `GZIPDecompressedStream`, which expect `bytes`.
fixEncode your string data to bytes before passing it to the stream. Example: `my_string.encode('utf-8')`. TypeError: 'AsyncGZIPCompressedStream' object is not async iterable
Trying to iterate an asynchronous stream object (`AsyncGZIPCompressedStream` or `AsyncGZIPDecompressedStream`) using a synchronous `for` loop.
fixEnsure you are using `async for chunk in stream:` within an `async def` function, and run the asynchronous code using `asyncio.run()`.
Upgrade
Version history
1.2.0latest on PyPI · released Dec 8, 2021
Audit
Dependencies
No dependency data recorded yet.