Caching &amp; Redis

import time
import threading
from collections import defaultdict


class CoalescingExecutor:
    def __init__(self):
        self._locks = defaultdict(threading.Lock)
        self._in_flight = {}

    def execute(self, key, func):
        with self._locks[key]:
            if key in self._in_flight:
                re…

caching write-through threading

How to Implement a Write-Through Cache in Python with a Mock Database

A thread-safe write-through cache that updates both cache and mock database atomically, computing values only after a successful write to the database.

import threading
import time
import random


class WriteThroughCache:
    def __init__(self):
        self.cache = {}
        self.db = {}
        self.lock = threading.Lock()

    def write(self, key, value):
        with self.lock:
            # Simulate slow database write
            time.sleep(random.uniform(0.01…

12 0 Open

How to Iterate Redis Keys with SCAN in Python

Iterate all Redis keys matching a pattern using the SCAN command with a mock client to simulate pagination.

redis scan keys

import redis

def scan_keys(client, pattern="*", count=10):
    keys = []
    cursor = 0
    while True:
        cursor, batch = client.scan(cursor=cursor, match=pattern, count=count)
        keys.extend(batch)
        if cursor == 0:
            break
    return keys

if __name__ == "__main__":
    # Mock Redis clien…

How to Mock a Redis Transaction with MULTI/EXEC in Python

A minimal in-memory mock of Redis MULTI/EXEC transactions that queues commands and applies them atomically on EXEC.

redis mock transactions

class RedisTransactionMock:
    def __init__(self):
        self.data = {}
        self.queue = []
        self.in_transaction = False

    def multi(self):
        self.in_transaction = True
        self.queue = []
        return "OK"

    def set(self, key, value):
        if self.in_transaction:
            self.qu…

14 0 Open

How to Use Redis HSET and HGET in Python

This code demonstrates how to store and retrieve hash data in Redis using Python's redis library with HSET, HGET, HGETALL, and HDEL commands.

redis hset hget

import redis

# Connect to Redis (adjust host/port as needed)
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# Clear any existing data for demonstration
r.delete('user:1')

# HSET - Store a hash
r.hset('user:1', mapping={'name': 'Alice', 'age': 30, 'city': 'New York'})

# HGET - Retrieve a …

12 0 Open

lru_cache caching functools

How to Use lru_cache in Python for Cache-on-Miss Population

Demonstrates lru_cache to automatically populate cache on a miss and serve subsequent calls from cache, with cache info stats.

from functools import lru_cache

@lru_cache(maxsize=None)
def fetch_user(user_id):
    """Simulates a slow database fetch."""
    print(f"Cache miss: fetching user {user_id} from database")
    return {"id": user_id, "name": f"User {user_id}"}

if __name__ == "__main__":
    user = fetch_user(1)
    print(f"First call…

lru_cache memoization functools

How to memoize a function in Python with lru_cache

Use functools.lru_cache to memoize a recursive Fibonacci function, caching results for a fixed number of calls to avoid repeated computation.

from functools import lru_cache

@lru_cache(maxsize=128)
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

if __name__ == "__main__":
    for i in range(10):
        print(f"fib({i}) = {fibonacci(i)}")
    print(f"Cache info: {fibonacci.cache_info()}")

13 0 Open

How to use Redis MGET MSET pipeline in Python

Store multiple keys atomically and read them efficiently with Redis MSET/MGET, then batch commands with a pipeline to cut round trips.

redis mget mset

import redis  # v4.x+ required

r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

# Sample data to store
r.flushdb()
data = {"name": "Alice", "age": "30", "city": "Berlin"}

# MSET: store multiple key-value pairs in one command
r.mset(data)

# MGET: fetch multiple keys in one round trip
keys =…

Implement a Multi-Level Cache with L1 Memory and L2 Redis in Python

This code implements a simple multi-level cache with an in-process L1 cache (via functools.lru_cache) and a mock Redis L2 cache with TTL, falling back to a slow computation on misses.

cache redis lru_cache

import time
from functools import lru_cache


class MockRedis:
    def __init__(self):
        self.store = {}

    def get(self, key):
        return self.store.get(key, None)

    def set(self, key, value, ttl=5):
        self.store[key] = (value, time.time() + ttl)

    def get_ttl(self, key):
        value, expiry…

Implement a TTL cache with a mock clock in Python

This code creates a simple TTL cache that stores values with an expiration timestamp and allows injecting a mock time function to test expiry behavior deterministically.

cache ttl mocking

import time
from functools import wraps

class TTLCache:
    def __init__(self, ttl_seconds):
        self.ttl = ttl_seconds
        self.cache = {}
        self._now = time.time

    def set_mock_time(self, mock_time_fn):
        """Inject a mock time function for testing TTL expiry."""
        self._now = mock_time_…