Cache Compression: Reducing Memory Usage in Cached Data

import redis
import json
import gzip

r = redis.Redis(decode_responses=True)

def compress_and_cache(key, data, ttl=3600, compression_level=6):
    """Compress data with gzip and store in Redis."""
    json_data = json.dumps(data)
    compressed = gzip.compress(json_data.encode(), compresslevel=compression_level)
    r.set(key, compressed)
    r.expire(key, ttl)
    return len(json_data), len(compressed)

def get_and_decompress(key):
    """Retrieve and decompress data from Redis."""
    compressed = r.get(key)
    if compressed is None:
        return None
    decompressed = gzip.decompress(compressed)
    return json.loads(decompressed)

large_report = {
    "title": "Monthly Analytics Report",
    "data": ["metric_value_" + str(i) for i in range(1000)],
    "metadata": {"generated": "2026-06-28", "version": "3.2.1"}
}

original_size, compressed_size = compress_and_cache(
    "report:monthly", large_report, ttl=3600
)
print(f"Original size: {original_size} bytes")
print(f"Compressed size: {compressed_size} bytes")
print(f"Compression ratio: {compressed_size / original_size:.1%}")

fetched = get_and_decompress("report:monthly")
print(f"Fetched title: {fetched['title']}")

import lz4.frame
import redis
import json

r = redis.Redis(decode_responses=True)

class LZ4Cache:
    def __init__(self, compression_level=0):
        self.compression_level = compression_level

    def set(self, key, data, ttl=3600):
        json_data = json.dumps(data)
        compressed = lz4.frame.compress(
            json_data.encode(),
            compression_level=self.compression_level
        )
        r.set(key, compressed)
        r.expire(key, ttl)
        return len(json_data), len(compressed)

    def get(self, key):
        compressed = r.get(key)
        if compressed is None:
            return None
        decompressed = lz4.frame.decompress(compressed)
        return json.loads(decompressed)

cache = LZ4Cache()

small_payload = {"status": "ok", "count": 42, "items": ["a", "b", "c"]}
orig, comp = cache.set("api:status", small_payload)
print(f"Small payload: {orig} -> {comp} bytes")

large_payload = {"data": ["x" * 1000 for _ in range(500)]}
orig, comp = cache.set("api:large", large_payload)
print(f"Large payload: {orig} -> {comp} bytes")
print(f"Ratio: {comp / orig:.1%}")

import zstandard as zstd
import redis
import json

r = redis.Redis(decode_responses=True)

class SmartCompressionCache:
    def __init__(self, min_compress_size=1024):
        self.min_compress_size = min_compress_size
        self.compressor = zstd.ZstdCompressor(level=3)
        self.decompressor = zstd.ZstdDecompressor()

    def set(self, key, data, ttl=3600):
        json_data = json.dumps(data)
        raw_size = len(json_data)

        if raw_size >= self.min_compress_size:
            compressed = self.compressor.compress(json_data.encode())
            r.set(f"{key}:z", compressed)
            r.expire(f"{key}:z", ttl)
            return {"size": raw_size, "compressed": len(compressed), "method": "zstd"}
        else:
            r.set(key, json_data)
            r.expire(key, ttl)
            return {"size": raw_size, "compressed": raw_size, "method": "none"}

    def get(self, key):
        compressed = r.get(f"{key}:z")
        if compressed is not None:
            data = self.decompressor.decompress(compressed)
            return json.loads(data)
        raw = r.get(key)
        if raw is not None:
            return json.loads(raw)
        return None

cache = SmartCompressionCache(min_compress_size=500)

small = {"msg": "ok"}
large = {"data": "x" * 5000}

result_small = cache.set("k:small", small)
print(f"Small: {result_small}")

result_large = cache.set("k:large", large)
print(f"Large: {result_large}")

print(f"Get small: {cache.get('k:small')}")
print(f"Get large keys: {list(cache.get('k:large').keys())}")