How to implement Redlock distributed lock in Python

import time
import random
import threading
from dataclasses import dataclass


@dataclass
class MockRedisLock:
    """Simple mock of a Redis lock instance."""
    name: str
    key: str
    ttl: int
    acquired: bool = False
    expires_at: float = 0.0

    def acquire(self, sleep_fn=time.sleep):
        """Try to acquire the lock. Returns True if successful."""
        if self.acquired and time.time() < self.expires_at:
            return False
        if random.random() < 0.3:  # simulate network failure
            return False
        self.acquired = True
        self.expires_at = time.time() + self.ttl
        sleep_fn(0.001)  # simulate round-trip
        return True

    def release(self):
        """Release the lock."""
        self.acquired = False


def acquire_redlock(instances, resource, ttl, majority=2):
    """Redlock: acquire lock from majority of instances."""
    locks = []
    for inst in instances:
        lock = MockRedisLock(inst, resource, ttl)
        if lock.acquire():
            locks.append(lock)
    if len(locks) >= majority:
        return locks
    for lock in locks:
        lock.release()
    return None


def simulate_debate():
    """Run a scenario showing Redlock's majority behavior."""
    instances = ["redis-1", "redis-2", "redis-3", "redis-4", "redis-5"]
    ttl = 5
    resource = "shared-resource"

    # Attempt 1: normal success
    locks = acquire_redlock(instances, resource, ttl, majority=3)
    status = "ACQUIRED" if locks else "FAILED"
    count = len(locks) if locks else 0
    print(f"Attempt 1: {status} (locks={count})")

    # Release all locks
    for lock in locks or []:
        lock.release()

    # Attempt 2: simulate bad network (lower success rate)
    random.seed(42)  # deterministic outcome
    locks = acquire_redlock(instances, resource, ttl, majority=3)
    status = "ACQUIRED" if locks else "FAILED"
    count = len(locks) if locks else 0
    print(f"Attempt 2 (bad network): {status} (locks={count})")


if __name__ == "__main__":
    simulate_debate()