(self, *, ignore_not_found=False)
| 210 | raise LockTimeout(str(self.store)) |
| 211 | |
| 212 | def release(self, *, ignore_not_found=False): |
| 213 | self.last_refresh_dt = None |
| 214 | locks = self._find_locks(only_mine=True) |
| 215 | if not locks: |
| 216 | if ignore_not_found: |
| 217 | logger.debug("LOCK-RELEASE: trying to release the lock, but none was found.") |
| 218 | return |
| 219 | else: |
| 220 | raise NotLocked(str(self.store)) |
| 221 | assert len(locks) == 1 |
| 222 | lock = locks[0] |
| 223 | logger.debug(f"LOCK-RELEASE: releasing lock: {lock}.") |
| 224 | self._delete_lock(lock["key"], ignore_not_found=True, update_last_refresh=True) |
| 225 | |
| 226 | def got_exclusive_lock(self): |
| 227 | locks = self._find_locks(only_mine=True, only_exclusive=True) |
no test coverage detected