Simple Compression Operations

def compress(data: bytes, level: int = 3) -> bytes:
    """
    Compress source data using the zstd compression format.

    This performs one-shot compression using basic/default compression
    settings.

    Parameters:
    - data: bytes-like object containing data to compress
    - level: int, compression level (1-22, default 3)

    Returns:
    bytes: Compressed data
    """

import zstandard as zstd

original = b"Hello, World! This is some data to compress."
compressed = zstd.compress(original, level=5)
print(f"Original: {len(original)} bytes, Compressed: {len(compressed)} bytes")

def decompress(data: bytes, max_output_size: int = 0) -> bytes:
    """
    Decompress a zstd frame into its original data.

    This performs one-shot decompression using basic/default compression
    settings.

    Parameters:
    - data: bytes-like object containing compressed data
    - max_output_size: int, maximum expected output size (0 for no limit)

    Returns:
    bytes: Decompressed data
    """

import zstandard as zstd

# Assuming 'compressed' contains zstd-compressed data
decompressed = zstd.decompress(compressed)
print(f"Decompressed: {decompressed}")

# With size limit for safety
decompressed = zstd.decompress(compressed, max_output_size=1024*1024)  # 1MB limit

def open(
    filename: Union[bytes, str, os.PathLike, BinaryIO],
    mode: str = "rb",
    cctx: Optional[ZstdCompressor] = None,
    dctx: Optional[ZstdDecompressor] = None,
    encoding: Optional[str] = None,
    errors: Optional[str] = None,
    newline: Optional[str] = None,
    closefd: bool = True
):
    """
    Create a file object with zstd (de)compression.

    The object returned from this function will be a
    ZstdDecompressionReader if opened for reading in binary mode,
    a ZstdCompressionWriter if opened for writing in binary mode,
    or an io.TextIOWrapper if opened for reading or writing in text mode.

    Parameters:
    - filename: Union[bytes, str, os.PathLike, BinaryIO], file path or file-like object
    - mode: str, file open mode ('rb', 'wb', 'rt', 'wt', etc.)
    - cctx: Optional[ZstdCompressor], compressor instance for compression
    - dctx: Optional[ZstdDecompressor], decompressor instance for decompression
    - encoding: Optional[str], text encoding for text mode
    - errors: Optional[str], error handling mode for text mode
    - newline: Optional[str], newline handling for text mode
    - closefd: bool, whether to close file descriptor when done (default: True)

    Returns:
    File-like object with compression/decompression
    """

import zstandard as zstd

# Write compressed data to file
with zstd.open('data.zst', 'wb') as f:
    f.write(b"This data will be compressed automatically")
    f.write(b"Multiple writes are supported")

# Read compressed data from file  
with zstd.open('data.zst', 'rb') as f:
    data = f.read()
    print(f"Read: {data}")

# Text mode with encoding
with zstd.open('text.zst', 'wt', encoding='utf-8') as f:
    f.write("This is text that will be compressed")
    f.write("Unicode characters: 🚀 ñ ü")

with zstd.open('text.zst', 'rt', encoding='utf-8') as f:
    text = f.read()
    print(f"Text: {text}")

# Using custom compressor/decompressor
compressor = zstd.ZstdCompressor(level=10)
with zstd.open('high-compression.zst', 'wb', cctx=compressor) as f:
    f.write(b"This will be compressed at level 10")