Release the lock only if owned by this instance. Returns True if the lock was successfully released.
(self)
| 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. |
| 63 | lua_script = """ |
| 64 | if redis.call("get", KEYS[1]) == ARGV[1] then |
| 65 | return redis.call("del", KEYS[1]) |
| 66 | else |
| 67 | return 0 |
| 68 | end |
| 69 | """ |
| 70 | release_lock = self.redis_client.register_script(lua_script) |
| 71 | result = release_lock(keys=[self.lock_key], args=[self.lock_token]) |
| 72 | return result == 1 |
| 73 | |
| 74 | # Example usage (for testing purposes only): |
| 75 | if __name__ == "__main__": |
no outgoing calls
no test coverage detected