Attempt to acquire the lock. Returns True if successful.
(self)
| 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 | """ |
no test coverage detected