(self, key, *, ignore_not_found=False, update_last_refresh=False)
| 110 | return key |
| 111 | |
| 112 | def _delete_lock(self, key, *, ignore_not_found=False, update_last_refresh=False): |
| 113 | logger.debug(f"LOCK-DELETE: deleting lock from store. key: {key}.") |
| 114 | try: |
| 115 | self.store.delete(f"locks/{key}") |
| 116 | except ObjectNotFound: |
| 117 | if not ignore_not_found: |
| 118 | raise |
| 119 | finally: |
| 120 | if update_last_refresh: |
| 121 | self.last_refresh_dt = None |
| 122 | |
| 123 | def _is_our_lock(self, lock): |
| 124 | return self.id == (lock["hostid"], lock["processid"], lock["threadid"]) |
no test coverage detected