Module

cache.compression

Cache compression utilities using Zstandard (PEP 784).

Python 3.14+ stdlib compression.zstd module for cache file compression. Provides transparent compression/decompression with backward compatibility.

Performance (from spike benchmarks):

  • Compression ratio: 12-14x on typical cache files
  • Size savings: 92-93%
  • Compress time: <1ms for typical files
  • Decompress time: <0.3ms for typical files

Related:

  • plan/active/rfc-zstd-cache-compression.md - RFC with benchmarks
  • bengal/cache/cache_store.py - Uses these utilities
  • bengal/cache/build_cache.py - Uses these utilities
View source
7 Functions

Functions

save_compressed
Save data as compressed JSON (.json.zst).
3 int 
def save_compressed(data: dict[str, Any], path: Path, level: int = COMPRESSION_LEVEL) -> int

Save data as compressed JSON (.json.zst).

Parameters 3

Name Type Default Description
data dict[str, Any] —

Dictionary to serialize
path Path —

Output path (should end in .json.zst)
level int COMPRESSION_LEVEL

Compression level 1-22 (default: 3)

Returns

int —

Number of bytes written (compressed size)

load_compressed
Load compressed JSON (.json.zst).
1 dict[str, Any] 
def load_compressed(path: Path) -> dict[str, Any]

Load compressed JSON (.json.zst).

Parameters 1

Name Type Default Description
path Path —

Path to compressed cache file

Returns

dict[str, Any] —

Deserialized dictionary

detect_format
Detect cache file format by extension.
1 str 
def detect_format(path: Path) -> str

Detect cache file format by extension.

Parameters 1

Name Type Default Description
path Path —

Path to cache file

Returns

str —

"zstd" for .json.zst files, "json" for .json files, "unknown" otherwise

get_compressed_path
Get the compressed path for a JSON cache file.
1 Path 
def get_compressed_path(json_path: Path) -> Path

Get the compressed path for a JSON cache file.

Parameters 1

Name Type Default Description
json_path Path —

Original JSON path (e.g., .bengal/cache.json)

Returns

Path —

Compressed path (e.g., .bengal/cache.json.zst)

get_json_path
Get the JSON path for a compressed cache file.
1 Path 
def get_json_path(compressed_path: Path) -> Path

Get the JSON path for a compressed cache file.

Parameters 1

Name Type Default Description
compressed_path Path —

Compressed path (e.g., .bengal/cache.json.zst)

Returns

Path —

JSON path (e.g., .bengal/cache.json)

load_auto
Load cache file with automatic format detection. Tries compressed format first (.json.zst), falls …
1 dict[str, Any] 
def load_auto(path: Path) -> dict[str, Any]

Load cache file with automatic format detection.

Tries compressed format first (.json.zst), falls back to JSON (.json). This enables seamless migration from uncompressed to compressed caches.

Parameters 1

Name Type Default Description
path Path —

Base path (without .zst extension)

Returns

dict[str, Any] —

Deserialized dictionary

migrate_to_compressed
Migrate an uncompressed JSON cache file to compressed format.
2 Path | None 
def migrate_to_compressed(json_path: Path, remove_original: bool = True) -> Path | None

Migrate an uncompressed JSON cache file to compressed format.

Parameters 2

Name Type Default Description
json_path Path —

Path to uncompressed JSON file
remove_original bool True

Whether to remove the original JSON file after migration

Returns

Path | None —

Path to compressed file, or None if migration failed/not needed