Python Redis WATCH optimistic lock mock

import time
import threading


class MockRedis:
    def __init__(self):
        self.data = {}
        self.watched = {}
        self.lock = threading.Lock()

    def get(self, key):
        return self.data.get(key)

    def set(self, key, value):
        self.data[key] = value

    def watch(self, *keys):
        with self.lock:
            for key in keys:
                self.watched[key] = self.data.get(key)

    def multi(self):
        return MockTransaction(self)

    def execute(self, transaction):
        with self.lock:
            for key, expected in self.watched.items():
                if self.data.get(key) != expected:
                    return None
            for key, value in transaction.operations:
                self.data[key] = value
            self.watched.clear()
            return transaction.operations


class MockTransaction:
    def __init__(self, redis):
        self.redis = redis
        self.operations = []

    def set(self, key, value):
        self.operations.append((key, value))


def transfer(redis, from_key, to_key, amount):
    redis.watch(from_key)
    balance = redis.get(from_key) or 0
    if balance < amount:
        redis.watched.clear()
        return False
    tx = redis.multi()
    tx.set(from_key, balance - amount)
    tx.set(to_key, (redis.get(to_key) or 0) + amount)
    return redis.execute(tx) is not None


if __name__ == "__main__":
    r = MockRedis()
    r.set("alice", 100)
    r.set("bob", 50)

    r.watch("alice")
    r.set("alice", 80)  # simulate concurrent modification
    result = transfer(r, "alice", "bob", 30)
    print(f"Transfer success: {result}")
    print(f"alice={r.get('alice')}, bob={r.get('bob')}")