Cache Stampede Prevention with SingleFlight in Python

import threading
import time
from functools import wraps


class SingleFlight:
    def __init__(self):
        self._lock = threading.Lock()
        self._inflight = None

    def do(self, key, fn):
        with self._lock:
            if self._inflight is not None:
                return self._inflight[1]
            result_holder = []
            done = threading.Event()

            def runner():
                try:
                    result_holder.append(fn())
                finally:
                    done.set()
                    with self._lock:
                        self._inflight = None

            self._inflight = (key, done)
            t = threading.Thread(target=runner)
            t.start()

        done.wait()
        return result_holder[0]


def mock_cache_and_stampede():
    cache = {"value": None}
    sf = SingleFlight()

    def expensive_compute():
        time.sleep(0.2)
        return 42

    def get_value():
        if cache["value"] is None:
            print("Cache miss, computing...")
            value = sf.do("data", expensive_compute)
            cache["value"] = value
        return cache["value"]

    threads = [threading.Thread(target=lambda: get_value()) for _ in range(5)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print(f"Final cached value: {cache['value']}")
    print(f"Only one expensive computation occurred (singleflight).")


if __name__ == "__main__":
    mock_cache_and_stampede()