Unconditionally remove all holders for a lock. Args: lock_name: Name identifying the lock Returns: True if any locks were removed
(self, lock_name: str)
| 308 | return removed |
| 309 | |
| 310 | def force_break(self, lock_name: str) -> bool: |
| 311 | """Unconditionally remove all holders for a lock. |
| 312 | |
| 313 | Args: |
| 314 | lock_name: Name identifying the lock |
| 315 | |
| 316 | Returns: |
| 317 | True if any locks were removed |
| 318 | """ |
| 319 | conn = self._get_connection() |
| 320 | try: |
| 321 | cursor = conn.cursor() |
| 322 | cursor.execute( |
| 323 | "DELETE FROM lock_holders WHERE lock_name = ?", |
| 324 | (lock_name,), |
| 325 | ) |
| 326 | deleted = cursor.rowcount > 0 |
| 327 | conn.commit() |
| 328 | return deleted |
| 329 | finally: |
| 330 | conn.close() |
| 331 | |
| 332 | def list_all_locks(self) -> list[dict[str, Any]]: |
| 333 | """List all lock records. |