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.

Medium Python 3.9+ Aug 9, 2026 Caching & Redis 14 views 0 copies

redis pipeline mock testing caching

Python code

64 lines

Python 3.9+

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":
                self.data[args[0]] = args[1]
                results.append(True)
            elif op == "GET":
                results.append(self.data.get(args[0]))
            elif op == "DELETE":
                results.append(self.data.pop(args[0], None) is not None)
        return results


class MockPipeline:
    def __init__(self, client):
        self.client = client
        self.commands = []

    def set(self, key, value):
        self.commands.append(("SET", key, value))
        return self

    def get(self, key):
        self.commands.append(("GET", key))
        return self

    def delete(self, key):
        self.commands.append(("DELETE", key))
        return self

    def execute(self):
        results = self.client.execute(self.commands)
        self.commands = []
        return results


if __name__ == "__main__":
    client = MockRedis()
    pipe = client.pipeline()
    pipe.set("name", "alice")
    pipe.set("age", "30")
    pipe.get("name")
    pipe.get("missing_key")
    pipe.delete("age")

    start = time.time()
    results = pipe.execute()
    elapsed = time.time() - start

    print("Results:", results)
    print(f"Pipeline executed in {elapsed:.6f} seconds")
    print("Final data:", client.data)

Output

stdout

Results: [True, True, 'alice', None, True]
Pipeline executed in 0.000023 seconds
Final data: {'name': 'alice'}

How it works

This mock mimics Redis pipelining by staging commands in a list instead of sending them immediately. The execute method flushes all buffered commands in one call, just like real Redis pipelines. Each command is simulated against a plain dictionary, keeping the code dependency-free and fast. The MockRedis class acts as the client and contains the actual data store, while MockPipeline handles command buffering and chainable methods. This pattern is ideal for unit tests where you want predictable behavior without network overhead.

Common mistakes

Forgetting to call `pipe.execute()` to actually run the buffered commands
Not clearing the command buffer after execution, causing stale commands to run again
Assuming operations are immediate — pipeline commands only affect data after `execute()`
Using real Redis in tests where network flakiness or latency would slow down the suite

Variations

Use `redis-py`'s built-in mock (`fakeredis`) for a more full-featured simulation
Implement a pipeline that returns deferred results and resolves them on `execute()`

Real-world use cases

Unit-testing cache-aside logic where multiple keys are read and written in a batch.
Simulating Redis pipeline behavior in local development without a Redis server.
Validating that your code batches commands correctly before deploying to production.

How to Mock Redis Pipeline Batch Commands in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Caching & Redis

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Caching & Redis

Keep learning

Tutorials

Quizzes