Caching &amp; Redis

from collections import OrderedDict
import time

class RedisMockDict:
    def __init__(self, ttl=None):
        self._data = OrderedDict()
        self._ttl = ttl  # default TTL in seconds, None = no expiry
        self._expiry = {}

    def set(self, key, value, ttl=None):
        """Set a key-value pair with optiona…

12 0 Open

How to Mock Redis EXPIRE, TTL, and PERSIST in Python

A lightweight in-memory MockRedis class that simulates Redis key expiration, TTL, and persist behavior for tests and local development.

redis mock ttl

import time

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

    def set(self, key, value):
        self._store[key] = value
        self._expiry.pop(key, None)
        return True

    def expire(self, key, ttl_seconds):
        if key not in self._store:
            retur…

How to Mock Redis Pipeline Batch Commands in Python

Create a lightweight MockRedis class that simulates Redis pipeline batching with SET, GET, and DELETE operations for testing without a live server.

redis pipeline mock

import redis
import time


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

    def pipeline(self):
        return MockPipeline(self)

    def execute(self, commands):
        results = []
        for cmd in commands:
            op, args = cmd[0], cmd[1:]
            if op == "SET":
                se…

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

redis optimistic-locking transactions

Python Redis WATCH optimistic lock mock

A MockRedis class that simulates WATCH/MULTI/EXEC transactions with optimistic locking to detect concurrent modifications before committing.

import time
import threading


class MockRedis:
    def __init__(self):
        self.data = {}
        self.watched = {}
        self.lock = threading.Lock()

    def get(self, key):
        return self.data.get(key)

    def set(self, key, value):
        self.data[key] = value

    def watch(self, *keys):
        wi…

12 0 Open

rate-limiting redis algorithms

Redis Leaky Bucket Rate Limiting Mock in Python

Simulates a Redis-backed leaky bucket rate limiter using a local class with continuous leaking and token capacity checks.

import time
from collections import deque


class LeakyBucket:
    def __init__(self, capacity, leak_rate):
        self.capacity = capacity
        self.leak_rate = leak_rate
        self.water = 0.0
        self.timestamp = time.time()
        self.history = deque()

    def allow(self):
        current = time.time(…