Mock Redis Distributed Lock in Python with SET NX EX

import time
import threading
import uuid
from typing import Optional


class RedisLockMock:
    """A minimal mock of Redis SET NX EX distributed lock semantics."""

    def __init__(self):
        self._store = {}  # key -> (value, expiry_epoch)

    def acquire(self, key: str, token: str, ttl_seconds: int) -> bool:
        now = time.time()
        # Clean expired lock if present
        if key in self._store and self._store[key][1] <= now:
            del self._store[key]
        if key not in self._store:
            self._store[key] = (token, now + ttl_seconds)
            return True
        return False

    def release(self, key: str, token: str) -> bool:
        entry = self._store.get(key)
        if entry and entry[0] == token:
            del self._store[key]
            return True
        return False

    def is_locked(self, key: str) -> bool:
        return key in self._store


def worker(lock: RedisLockMock, key: str, ttl: int, result: list):
    token = str(uuid.uuid4())
    if lock.acquire(key, token, ttl):
        result.append(f"acquired with {token[:8]}")
        time.sleep(0.1)
        lock.release(key, token)
        result.append("released")
    else:
        result.append(f"failed: {token[:8]}")


if __name__ == "__main__":
    lock = RedisLockMock()
    key = "resource:123"
    results = []

    t1 = threading.Thread(target=worker, args=(lock, key, 5, results))
    t2 = threading.Thread(target=worker, args=(lock, key, 5, results))
    t1.start()
    t2.start()
    t1.join()
    t2.join()

    print(results)
    print(f"lock still held: {lock.is_locked(key)}")