Release lock for a specific PID. Args: lock_name: Name identifying the lock owner_pid: PID of the process releasing the lock Returns: True if a lock was released, False if no lock was held
(self, lock_name: str, owner_pid: int)
| 179 | conn.close() |
| 180 | |
| 181 | def release(self, lock_name: str, owner_pid: int) -> bool: |
| 182 | """Release lock for a specific PID. |
| 183 | |
| 184 | Args: |
| 185 | lock_name: Name identifying the lock |
| 186 | owner_pid: PID of the process releasing the lock |
| 187 | |
| 188 | Returns: |
| 189 | True if a lock was released, False if no lock was held |
| 190 | """ |
| 191 | conn = self._get_connection() |
| 192 | try: |
| 193 | cursor = conn.cursor() |
| 194 | cursor.execute( |
| 195 | "DELETE FROM lock_holders WHERE lock_name = ? AND owner_pid = ?", |
| 196 | (lock_name, owner_pid), |
| 197 | ) |
| 198 | deleted = cursor.rowcount > 0 |
| 199 | conn.commit() |
| 200 | return deleted |
| 201 | finally: |
| 202 | conn.close() |
| 203 | |
| 204 | def is_held(self, lock_name: str) -> bool: |
| 205 | """Check if any holder exists for this lock. |
nothing calls this directly
no test coverage detected