types-zstd

import zstd from typing import Optional # Ensure python-zstandard is installed: pip install python-zstandard def compress_data(data: bytes, level: Optional[int] = None) -> bytes: """Compresses bytes using zstd.""" if level is not None: return zstd.compress(data, level) return zstd.compress(data) def decompress_data(compressed_data: bytes) -> bytes: """Decompresses bytes using zstd.""" return zstd.decompress(compressed_data) if __name__ == '__main__': original_data = b"This is some data to be compressed by zstd! " * 100 # Use type-checked functions compressed = compress_data(original_data, level=3) decompressed = decompress_data(compressed) print(f"Original size: {len(original_data)} bytes") print(f"Compressed size: {len(compressed)} bytes") print(f"Decompressed matches original: {original_data == decompressed}") # Example of using the streaming API (also type-checked) compressor = zstd.ZstdCompressor(level=1) decompressor = zstd.ZstdDecompressor() with compressor.stream_writer() as writer: writer.write(b"Hello ") writer.write(b"World!") compressed_stream = writer.read() with decompressor.stream_reader(compressed_stream) as reader: decompressed_stream = reader.read() print(f"Stream compression works: {decompressed_stream == b'Hello World!'}")