Caching &amp; Redis

import hashlib
import random

class BloomFilter:
    def __init__(self, size=100, num_hashes=3):
        self.size = size
        self.num_hashes = num_hashes
        self.bit_array = [0] * size

    def _hashes(self, item):
        result = []
        for i in range(self.num_hashes):
            hash_value = int(hash…

How to Implement a Negative Cache with TTL in Python

This code provides a TTL mock cache that stores negative results (cache misses) for a short time to reduce repeated lookups of missing keys.

cache ttl negative-cache

from time import time, sleep

class TTLMockCache:
    def __init__(self, ttl_seconds=5):
        self.ttl = ttl_seconds
        self.store = {}
        self.negative_cache = {}

    def get(self, key):
        now = time()
        if key in self.store:
            value, expires_at = self.store[key]
            if exp…

How to Mock Cache Tag Invalidation in Python

Use unittest.mock.patch with wraps to verify tagged cache entries are invalidated correctly.

unittest mock cache

import unittest
from unittest.mock import patch

def get_cached_data(cache, key):
    """Return data from cache if present and valid, else None."""
    if cache.get(key, {}).get("valid", False):
        return cache[key]["data"]
    return None

def invalidate_tag_mock(cache, tag):
    """Invalidate all cache entries …

How to Mock Redis Streams Consumer Groups in Python

Simulate Redis Streams producer and consumer group behavior in Python using a standalone mock class for testing and development.

redis streams mock

import time
import json
from collections import defaultdict

class RedisStreamMock:
    def __init__(self):
        self.streams = defaultdict(list)
        self.consumer_groups = defaultdict(dict)
        self.pending_entries = defaultdict(list)

    def xadd(self, stream, fields):
        entry_id = f"{time.time_ns(…

Caching & Redis easy

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 =…

15 0 Open

redis rate-limit sliding-window

Redis-inspired sliding window rate limiter in Python

A pure-Python sliding window rate limiter using a deque of timestamps, mock-ready for Redis-backed production limits.

import time
from collections import deque


class SlidingWindowRateLimiter:
    def __init__(self, max_requests: int, window_seconds: int) -> None:
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests: dict[str, deque] = {}

    def is_allowed(self, client_id: str…

Refresh Proactive TTL Renewal in Python

This snippet implements a proactive TTL renewal pattern that refreshes a cache expiration before it lapses, using a mock counter to track renewals.

caching ttl renewal

redis caching cache-aside

import time
from datetime import datetime, timezone

class TTLRenewer:
    def __init__(self, ttl_seconds=10, renew_at=0.5):
        self.ttl = ttl_seconds
        self.last_renewed = time.time()
        self.renew_threshold = ttl_seconds * renew_at
        self.renewals = 0

    def check_and_renew(self):
        if …

13 0 Open

Caching & Redis easy

Simple Redis Cache Helper in Python

Build a minimal Redis-backed cache with TTL, JSON serialization, and automated fetching to speed up repeated expensive lookups.