Get all holders for a lock with metadata. Args: lock_name: Name identifying the lock Returns: List of dicts with holder information
(self, lock_name: str)
| 223 | conn.close() |
| 224 | |
| 225 | def get_lock_info(self, lock_name: str) -> list[dict[str, Any]]: |
| 226 | """Get all holders for a lock with metadata. |
| 227 | |
| 228 | Args: |
| 229 | lock_name: Name identifying the lock |
| 230 | |
| 231 | Returns: |
| 232 | List of dicts with holder information |
| 233 | """ |
| 234 | conn = self._get_connection() |
| 235 | try: |
| 236 | cursor = conn.cursor() |
| 237 | cursor.execute( |
| 238 | "SELECT lock_name, owner_pid, lock_mode, operation, hostname, acquired_at " |
| 239 | "FROM lock_holders WHERE lock_name = ?", |
| 240 | (lock_name,), |
| 241 | ) |
| 242 | rows = cursor.fetchall() |
| 243 | return [ |
| 244 | { |
| 245 | "lock_name": r[0], |
| 246 | "owner_pid": r[1], |
| 247 | "lock_mode": r[2], |
| 248 | "operation": r[3], |
| 249 | "hostname": r[4], |
| 250 | "acquired_at": r[5], |
| 251 | } |
| 252 | for r in rows |
| 253 | ] |
| 254 | finally: |
| 255 | conn.close() |
| 256 | |
| 257 | def is_lock_stale(self, lock_name: str) -> bool: |
| 258 | """Check if ALL holders have dead PIDs. |