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.
Python code
57 linesclass 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.queue.append(("set", key, value))
return "QUEUED"
self.data[key] = value
return "OK"
def get(self, key):
if self.in_transaction:
self.queue.append(("get", key))
return "QUEUED"
return self.data.get(key)
def delete(self, key):
if self.in_transaction:
self.queue.append(("delete", key))
return "QUEUED"
self.data.pop(key, None)
return "OK"
def exec(self):
if not self.in_transaction:
return None
results = []
for command in self.queue:
if command[0] == "set":
self.data[command[1]] = command[2]
results.append("OK")
elif command[0] == "get":
results.append(self.data.get(command[1]))
elif command[0] == "delete":
self.data.pop(command[1], None)
results.append("OK")
self.in_transaction = False
self.queue = []
return results
if __name__ == "__main__":
r = RedisTransactionMock()
r.multi()
print(r.set("name", "Alice"))
print(r.get("name"))
print(r.set("age", 30))
result = r.exec()
print(result)
print(r.get("name"))
Output
QUEUED
QUEUED
QUEUED
['OK', 'Alice', 'OK']
Alice
How it works
The mock starts a transaction with multi, which sets in_transaction to True and clears the command queue. Each subsequent command checks the flag; if inside a transaction, it appends the command to the queue and returns 'QUEUED', mimicking true Redis behavior. exec replays the queued commands in order against the in-memory data store, collects their results, resets the transaction state, and returns the result list. This simulates the atomic batch semantics of real Redis transactions without needing a server or network I/O, making it ideal for unit tests.
Common mistakes
- Forgetting to reset transaction state after exec, leaving in_transaction True for subsequent commands.
- Not checking in_transaction in get/set/delete, so commands execute immediately instead of queuing.
- Assuming exec returns a single value when it always returns a list of per-command results.
Variations
- Use a real Redis client with `pipeline(transaction=True)` for server-side atomicity.
- Implement WATCH support to detect concurrent modifications before EXEC.
Real-world use cases
- Unit testing application code that depends on Redis transactions without needing a live Redis instance.
- Simulating MULTI/EXEC behavior in CI environments to test logic deterministically and fast.
- Prototyping transaction flows during development without external service dependencies.
Sponsored
More from Caching & Redis
- Cache Asides in Python with a Read-Through Loader easy
- Cache Data in Redis with Python easy
- Cache Penetration Null Object Mock in Python medium
- Cache Stampede Prevention with SingleFlight in Python medium
- Cache Warming with Python: Preload Hot Keys easy
- Coalescing duplicate in-flight requests: one shared result for concurrent callers hard
Keep learning
Related tutorials and quizzes for this topic.