tessl/pypi-lz4
LZ4 Bindings for Python providing high-performance lossless data compression with frame and block format support
—
Pending
frame-format.mddocs/
Frame Format
Frame format compression provides production-ready LZ4 compression with interoperability guarantees, metadata support, and extensive configuration options. This is the recommended interface for most compression use cases.
Capabilities
Simple Compression Functions
High-level functions for straightforward compression and decompression operations.
def compress(data, **kwargs) -> bytes:
"""
Compress data using LZ4 frame format.
Args:
data (bytes-like): Data to compress
block_size (int, optional): Maximum block size (default: BLOCKSIZE_DEFAULT)
block_linked (bool, optional): Use block-linked compression (default: True)
compression_level (int, optional): Compression level 0-16 (default: 0)
content_checksum (bool, optional): Enable content checksum (default: False)
block_checksum (bool, optional): Enable block checksum (default: False, requires LZ4 >= 1.8.0)
auto_flush (bool, optional): Disable buffering (default: False)
source_size (int, optional): Size hint for uncompressed data (default: 0)
return_bytearray (bool, optional): Return bytearray instead of bytes (default: False)
Returns:
bytes: Compressed data in LZ4 frame format
"""
def decompress(data, **kwargs) -> bytes:
"""
Decompress LZ4 frame format data.
Args:
data (bytes-like): Compressed data in LZ4 frame format
return_bytearray (bool, optional): Return bytearray instead of bytes (default: False)
Returns:
bytes: Decompressed data
"""
def open(filename, mode, **kwargs) -> LZ4FrameFile:
"""
Open an LZ4 compressed file.
Args:
filename (str|bytes|PathLike|file): File path or file object
mode (str): File mode ('r', 'w', 'x', 'a', with optional 'b' or 't')
**kwargs: All LZ4FrameCompressor parameters for writing modes
source_size (int, optional): Total uncompressed size hint (default: 0)
return_bytearray (bool, optional): Return bytearray objects (default: False)
Returns:
LZ4FrameFile: File-like object for LZ4 compressed files
"""Low-level Context Management
Functions for manual frame compression with full control over the compression process.
def create_compression_context():
"""Create compression context for low-level operations."""
def compress_begin(context, **kwargs) -> bytes:
"""
Begin compression frame with specified options.
Args:
context: Compression context from create_compression_context()
**kwargs: Same parameters as compress() function
Returns:
bytes: Frame header
"""
def compress_chunk(context, data, **kwargs) -> bytes:
"""
Compress data chunk using existing context.
Args:
context: Active compression context
data (bytes-like): Data chunk to compress
**kwargs: Compression options
Returns:
bytes: Compressed chunk data
"""
def compress_flush(context, **kwargs) -> bytes:
"""
Flush compression context and finalize frame.
Args:
context: Active compression context
**kwargs: Flush options
Returns:
bytes: Final frame data and footer
"""
def create_decompression_context():
"""Create decompression context for low-level operations."""
def reset_decompression_context(context) -> None:
"""
Reset decompression context (requires LZ4 >= 1.8.0).
Args:
context: Decompression context to reset
"""
def decompress_chunk(context, data, **kwargs) -> tuple[bytes, int, bool]:
"""
Decompress data chunk using existing context.
Args:
context: Active decompression context
data (bytes-like): Compressed data chunk
**kwargs: Decompression options
Returns:
tuple: (decompressed_data, bytes_consumed, frame_complete)
"""
def get_frame_info(data, **kwargs) -> dict:
"""
Get information about LZ4 frame.
Args:
data (bytes-like): Frame data (at least frame header)
**kwargs: Options for frame info extraction
Returns:
dict: Frame information including block size, checksums, etc.
"""Constants
Block size and compression level constants for configuration.
BLOCKSIZE_DEFAULT: int # Default block size (equals BLOCKSIZE_MAX64KB)
BLOCKSIZE_MAX64KB: int # 64 KB maximum block size
BLOCKSIZE_MAX256KB: int # 256 KB maximum block size
BLOCKSIZE_MAX1MB: int # 1 MB maximum block size
BLOCKSIZE_MAX4MB: int # 4 MB maximum block size
COMPRESSIONLEVEL_MIN: int # Minimum compression level (0)
COMPRESSIONLEVEL_MINHC: int # Minimum high-compression level (3)
COMPRESSIONLEVEL_MAX: int # Maximum compression level (16)Incremental Compression
Class for incremental compression with full configuration control.
class LZ4FrameCompressor:
"""Incremental frame compression with extensive configuration options."""
def __init__(
self,
block_size: int = BLOCKSIZE_DEFAULT,
block_linked: bool = True,
compression_level: int = 0,
content_checksum: bool = False,
block_checksum: bool = False,
auto_flush: bool = False,
return_bytearray: bool = False
):
"""
Initialize incremental compressor.
Args:
block_size: Maximum block size for compression
block_linked: Use block-linked compression for better ratio
compression_level: Compression level 0-16 (0=fastest, 16=best)
content_checksum: Enable content checksum verification
block_checksum: Enable per-block checksums (requires LZ4 >= 1.8.0)
auto_flush: Disable buffering for immediate output
return_bytearray: Return bytearray instead of bytes
"""
def begin(self, source_size: int = 0) -> bytes:
"""
Begin compression frame.
Args:
source_size: Size hint for total uncompressed data
Returns:
bytes: Frame header
"""
def compress(self, data: bytes) -> bytes:
"""
Compress data chunk.
Args:
data: Data chunk to compress
Returns:
bytes: Compressed chunk data
"""
def flush(self) -> bytes:
"""
Finish compression and return final frame data.
Returns:
bytes: Final compressed data and frame footer
"""
def reset(self) -> None:
"""Reset compressor state for reuse."""
def has_context(self) -> bool:
"""Check if compression context exists."""
def started(self) -> bool:
"""Check if compression frame has been started."""
def __enter__(self):
"""Context manager entry."""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Context manager exit."""Incremental Decompression
Class for incremental decompression with state management.
class LZ4FrameDecompressor:
"""Incremental frame decompression with state tracking."""
def __init__(self, return_bytearray: bool = False):
"""
Initialize incremental decompressor.
Args:
return_bytearray: Return bytearray instead of bytes
"""
@property
def eof(self) -> bool:
"""True if end-of-stream reached."""
@property
def unused_data(self) -> bytes:
"""Data after compressed stream."""
@property
def needs_input(self) -> bool:
"""True if more input data needed for decompression."""
def decompress(self, data: bytes, max_length: int = -1) -> bytes:
"""
Decompress data chunk.
Args:
data: Compressed data chunk
max_length: Maximum bytes to decompress (-1 for unlimited)
Returns:
bytes: Decompressed data
"""
def reset(self) -> None:
"""Reset decompressor state for reuse."""
def __enter__(self):
"""Context manager entry."""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Context manager exit."""File Interface
File-like object providing standard Python file interface for LZ4 compressed files.
class LZ4FrameFile:
"""File-like object for LZ4 compressed files with standard Python file interface."""
def __init__(
self,
filename,
mode: str,
block_size: int = BLOCKSIZE_DEFAULT,
block_linked: bool = True,
compression_level: int = 0,
content_checksum: bool = False,
block_checksum: bool = False,
auto_flush: bool = False,
source_size: int = 0,
return_bytearray: bool = False
):
"""
Initialize LZ4 file object.
Args:
filename: File path or file object
mode: File mode ('r', 'w', 'x', 'a', with optional 'b' or 't')
block_size: Block size for compression
block_linked: Use block-linked compression
compression_level: Compression level 0-16
content_checksum: Enable content checksum
block_checksum: Enable block checksums
auto_flush: Disable buffering
source_size: Total uncompressed size hint
return_bytearray: Return bytearray objects
"""
@property
def closed(self) -> bool:
"""True if file is closed."""
def read(self, size: int = -1) -> bytes:
"""Read uncompressed data."""
def read1(self, size: int = -1) -> bytes:
"""Read data avoiding multiple reads."""
def readall(self) -> bytes:
"""Read entire file."""
def readline(self, size: int = -1) -> bytes:
"""Read line from file."""
def write(self, data: bytes) -> int:
"""Write uncompressed data."""
def flush(self) -> None:
"""Flush write buffers."""
def seek(self, offset: int, whence: int = 0) -> int:
"""Seek to position in file."""
def tell(self) -> int:
"""Get current file position."""
def peek(self, size: int = -1) -> bytes:
"""Peek at buffered data without consuming it."""
def close(self) -> None:
"""Close file and flush buffers."""
def readable(self) -> bool:
"""Check if file is readable."""
def writable(self) -> bool:
"""Check if file is writable."""
def seekable(self) -> bool:
"""Check if file supports seeking."""
def fileno(self) -> int:
"""Get underlying file descriptor."""Usage Examples
Simple Compression
import lz4.frame
# Basic compression
data = b"Hello, World!" * 1000
compressed = lz4.frame.compress(data)
decompressed = lz4.frame.decompress(compressed)
# High compression with checksums
compressed = lz4.frame.compress(
data,
compression_level=9,
content_checksum=True
)Incremental Processing
import lz4.frame
# Incremental compression
with lz4.frame.LZ4FrameCompressor(compression_level=9) as compressor:
header = compressor.begin(source_size=len(total_data))
compressed_parts = [header]
for chunk in data_chunks:
compressed_parts.append(compressor.compress(chunk))
compressed_parts.append(compressor.flush())
compressed_data = b''.join(compressed_parts)
# Incremental decompression
with lz4.frame.LZ4FrameDecompressor() as decompressor:
decompressed_parts = []
for chunk in compressed_chunks:
part = decompressor.decompress(chunk)
if part:
decompressed_parts.append(part)
if decompressor.eof:
break
decompressed_data = b''.join(decompressed_parts)File Operations
import lz4.frame
# Writing compressed files
with lz4.frame.open('data.lz4', 'wb', compression_level=9) as f:
f.write(b"Large amounts of data...")
f.write(b"More data...")
# Reading compressed files
with lz4.frame.open('data.lz4', 'rb') as f:
content = f.read()
# Text mode with encoding
with lz4.frame.open('text.lz4', 'wt', encoding='utf-8') as f:
f.write("Unicode text content")
with lz4.frame.open('text.lz4', 'rt', encoding='utf-8') as f:
text = f.read()Install with Tessl CLI
npx tessl i tessl/pypi-lz4