Cache Batch Operations: Efficient Bulk Read and Write with MSET, MGET, and Pipelines

import redis

r = redis.Redis(decode_responses=True)

class BatchCache:
    def __init__(self, redis_client):
        self.r = redis_client

    def get_multi(self, keys):
        """Get multiple keys in one round-trip."""
        values = self.r.mget(keys)
        result = {}
        for key, value in zip(keys, values):
            if value is not None:
                result[key] = value
        return result

    def set_multi(self, mapping, ttl=3600):
        """Set multiple keys atomically."""
        pipe = self.r.pipeline(transaction=True)
        pipe.multi()
        for key, value in mapping.items():
            pipe.setex(key, ttl, value)
        results = pipe.execute()
        return {"set": len(mapping), "success": all(r is True for r in results)}

    def get_or_fetch_multi(self, keys, fetch_fn, ttl=3600):
        """Batch cache-aside: get cached, fetch missing, cache fetched."""
        cached = self.get_multi(keys)
        missing = [k for k in keys if k not in cached]

        if missing:
            fetched = fetch_fn(missing)
            for key, value in fetched.items():
                r.setex(key, ttl, value)
            cached.update(fetched)

        return cached

cache = BatchCache(r)

cache.set_multi({
    "batch:user:1": "Alice",
    "batch:user:2": "Bob",
    "batch:user:3": "Charlie",
}, ttl=300)

keys = ["batch:user:1", "batch:user:2", "batch:user:3", "batch:nonexistent"]
results = cache.get_multi(keys)
print(f"Batch GET: {len(results)} of {len(keys)} found")
for key, value in results.items():
    print(f"  {key}: {value}")

import redis
import time

r = redis.Redis(decode_responses=True)

class PipelineBatch:
    def __init__(self, redis_client):
        self.r = redis_client

    def benchmark(self, operation_count=1000):
        """Compare sequential vs pipelined performance."""
        keys = [f"bench:{i}" for i in range(operation_count)]

        sequential_start = time.perf_counter()
        for i, key in enumerate(keys):
            self.r.setex(key, 3600, f"value_{i}")
        sequential_time = time.perf_counter() - sequential_start

        pipe = self.r.pipeline()
        pipe_start = time.perf_counter()
        for i, key in enumerate(keys):
            pipe.setex(key, 3600, f"pipe_value_{i}")
        pipe.execute()
        pipe_time = time.perf_counter() - pipe_start

        return {
            "operation_count": operation_count,
            "sequential_seconds": round(sequential_time, 3),
            "pipeline_seconds": round(pipe_time, 3),
            "speedup": round(sequential_time / pipe_time, 1) if pipe_time > 0 else 0,
        }

    def batch_with_sizes(self, key_count=1000):
        """Compare different pipeline batch sizes."""
        base_keys = [f"batchsize:{i}" for i in range(key_count)]
        results = {}

        for batch_size in [10, 50, 100, 500, 1000]:
            start = time.perf_counter()
            pipe = self.r.pipeline()
            for i, key in enumerate(base_keys):
                pipe.setex(f"{key}:bs{batch_size}", 3600, f"val_{i}")
                if (i + 1) % batch_size == 0:
                    pipe.execute()
                    pipe = self.r.pipeline()
            if pipe.command_stack:
                pipe.execute()
            elapsed = time.perf_counter() - start
            results[batch_size] = round(elapsed, 3)

        return results

pipeline = PipelineBatch(r)

bench = pipeline.benchmark(operation_count=500)
print(f"Sequential: {bench['sequential_seconds']}s")
print(f"Pipeline: {bench['pipeline_seconds']}s")
print(f"Speedup: {bench['speedup']}x")

batch_results = pipeline.batch_with_sizes(500)
print(f"\nBatch size performance:")
for size, time_taken in batch_results.items():
    print(f"  Batch size {size:5d}: {time_taken}s")

import redis
import time

r = redis.Redis(decode_responses=True)

class BulkLoader:
    def __init__(self, redis_client, batch_size=100):
        self.r = redis_client
        self.batch_size = batch_size

    def load_data(self, items, ttl=3600):
        """Load a large number of items with reporting."""
        total = len(items)
        loaded = 0
        errors = 0
        start = time.time()

        pipe = self.r.pipeline()
        for i, (key, value) in enumerate(items):
            try:
                pipe.setex(key, ttl, value)
                loaded += 1
                if (i + 1) % self.batch_size == 0:
                    pipe.execute()
                    pipe = self.r.pipeline()
                    elapsed = time.time() - start
                    rate = loaded / elapsed if elapsed > 0 else 0
                    print(f"  Progress: {loaded}/{total} ({rate:.0f} keys/s)", end="\r")
            except Exception:
                errors += 1

        if pipe.command_stack:
            pipe.execute()

        elapsed = time.time() - start
        return {
            "total": total,
            "loaded": loaded,
            "errors": errors,
            "elapsed_seconds": round(elapsed, 2),
            "rate_per_second": round(loaded / elapsed) if elapsed > 0 else 0,
        }

    def load_from_generator(self, generator, ttl=3600):
        """Load data from a generator function."""
        items = list(generator)
        return self.load_data(items, ttl)

def generate_items(count):
    for i in range(count):
        yield (f"bulk:key:{i}", f"value_{i}")

loader = BulkLoader(r, batch_size=200)

items = list(generate_items(5000))
result = loader.load_data(items, ttl=3600)
print(f"\nBulk load complete:")
print(f"  Loaded: {result['loaded']:,} keys")
print(f"  Errors: {result['errors']}")
print(f"  Time: {result['elapsed_seconds']}s")
print(f"  Rate: {result['rate_per_second']:,} keys/s")