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.

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

caching write-through threading mock-database concurrency

Python code

41 lines

Python 3.9+

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, 0.05))
            self.db[key] = value
            # Write through to cache
            self.cache[key] = value
            return f"Wrote {key}={value} to cache and DB"

    def read(self, key):
        with self.lock:
            if key in self.cache:
                return self.cache[key]
            # Cache miss - read from DB
            value = self.db.get(key, None)
            if value is not None:
                self.cache[key] = value
            return value


if __name__ == "__main__":
    cache = WriteThroughCache()
    cache.write("user:1", {"name": "Alice", "age": 30})
    cache.write("user:2", {"name": "Bob", "age": 25})

    print("Cache content:", cache.cache)
    print("DB content:", cache.db)
    print("Read user:1 ->", cache.read("user:1"))
    print("Read user:3 (missing) ->", cache.read("user:3"))
    print("Cache after reads:", cache.cache)

Output

stdout

Cache content: {'user:1': {'name': 'Alice', 'age': 30}, 'user:2': {'name': 'Bob', 'age': 25}}
DB content: {'user:1': {'name': 'Alice', 'age': 30}, 'user:2': {'name': 'Bob', 'age': 25}}
Read user:1 -> {'name': 'Alice', 'age': 30}
Read user:3 (missing) -> None
Cache after reads: {'user:1': {'name': 'Alice', 'age': 30}, 'user:2': {'name': 'Bob', 'age': 25}}

How it works

The write method uses a threading.Lock to guarantee atomicity between the simulated database write and cache update, so the cache is never stale or partially updated. The read method checks the cache first and only falls back to the mock database on a miss, populating the cache with the fetched value. The random sleep simulates real-world I/O latency, making the locking necessary to prevent race conditions in concurrent systems. With the lock, the cache and database states stay consistent because every write goes through both layers in a single critical section.

Common mistakes

Forgetting to acquire the lock on read operations, which can return stale cache data while a write is in progress
Updating the cache before the database write succeeds, leaving the cache ahead of the source of truth on failure
Using a real database connection without mocking, making tests slow and non-deterministic

Variations

Use an LRU eviction policy for the cache to handle size limits and remove the least recently used entries
Batch write-through operations with a background flush for high-throughput scenarios instead of synchronous writes

Real-world use cases

Keeping an in-memory session store consistent with a backing user database in a web app.
Warming a product catalog cache while ensuring order processing always sees fresh inventory data.
Testing cache-layer integration without spinning up a full database, using an in-memory mock.

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

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