A persistent, auto-expiring lock that uses Redis. Usage: 1. Instantiate with a Redis client, a unique lock key (e.g. "lock:account:123"), and an optional timeout (in seconds). 2. Call acquire() to try to obtain the lock. 3. Optionally, periodically call refresh()
| 3 | import redis |
| 4 | |
| 5 | class PersistentLock: |
| 6 | """ |
| 7 | A persistent, auto-expiring lock that uses Redis. |
| 8 | |
| 9 | Usage: |
| 10 | 1. Instantiate with a Redis client, a unique lock key (e.g. "lock:account:123"), |
| 11 | and an optional timeout (in seconds). |
| 12 | 2. Call acquire() to try to obtain the lock. |
| 13 | 3. Optionally, periodically call refresh() to extend the lock's lifetime. |
| 14 | 4. When finished, call release() to free the lock. |
| 15 | """ |
| 16 | def __init__(self, redis_client: redis.Redis, lock_key: str, lock_timeout: int = 120): |
| 17 | """ |
| 18 | Initialize the lock. |
| 19 | |
| 20 | :param redis_client: An instance of redis.Redis. |
| 21 | :param lock_key: The unique key for the lock. |
| 22 | :param lock_timeout: Time-to-live for the lock in seconds. |
| 23 | """ |
| 24 | self.redis_client = redis_client |
| 25 | self.lock_key = lock_key |
| 26 | self.lock_timeout = lock_timeout |
| 27 | self.lock_token = None |
| 28 | self.has_lock = False |
| 29 | |
| 30 | def has_lock(self) -> bool: |
| 31 | return self.has_lock |
| 32 | |
| 33 | def acquire(self) -> bool: |
| 34 | """ |
| 35 | Attempt to acquire the lock. Returns True if successful. |
| 36 | """ |
| 37 | self.lock_token = str(uuid.uuid4()) |
| 38 | # Set the lock with NX (only if not exists) and EX (expire time) |
| 39 | result = self.redis_client.set(self.lock_key, self.lock_token, nx=True, ex=self.lock_timeout) |
| 40 | if result is not None: |
| 41 | self.has_lock = True |
| 42 | |
| 43 | return result is not None |
| 44 | |
| 45 | def refresh(self) -> bool: |
| 46 | """ |
| 47 | Refresh the lock's expiration time if this instance owns the lock. |
| 48 | Returns True if the expiration was successfully extended. |
| 49 | """ |
| 50 | current_value = self.redis_client.get(self.lock_key) |
| 51 | if current_value and current_value == self.lock_token: |
| 52 | self.redis_client.expire(self.lock_key, self.lock_timeout) |
| 53 | self.has_lock = False |
| 54 | return True |
| 55 | return False |
| 56 | |
| 57 | def release(self) -> bool: |
| 58 | """ |
| 59 | Release the lock only if owned by this instance. |
| 60 | Returns True if the lock was successfully released. |
| 61 | """ |
| 62 | # Use a Lua script for atomicity: only delete if the token matches. |
no outgoing calls
no test coverage detected