Find all process IDs that have open handles to files under the given path. Args: path: Path to check for locks (file or directory) max_processes: Maximum number of processes to check (default: 500) timeout: Maximum time in seconds to spend checking (default: 60.0) R
(
path: Path, max_processes: int = 500, timeout: float = 60.0
)
| 28 | |
| 29 | |
| 30 | def find_processes_locking_path( |
| 31 | path: Path, max_processes: int = 500, timeout: float = 60.0 |
| 32 | ) -> set[int]: |
| 33 | """Find all process IDs that have open handles to files under the given path. |
| 34 | |
| 35 | Args: |
| 36 | path: Path to check for locks (file or directory) |
| 37 | max_processes: Maximum number of processes to check (default: 500) |
| 38 | timeout: Maximum time in seconds to spend checking (default: 60.0) |
| 39 | |
| 40 | Returns: |
| 41 | Set of process IDs holding locks on the path |
| 42 | """ |
| 43 | if not is_psutil_available(): |
| 44 | print("Warning: psutil not available, cannot detect lock holders") |
| 45 | return set() |
| 46 | |
| 47 | if not path.exists(): |
| 48 | return set() |
| 49 | |
| 50 | # Normalize path for comparison |
| 51 | try: |
| 52 | normalized_path = path.resolve() |
| 53 | except (OSError, RuntimeError): |
| 54 | # Path might be in a bad state |
| 55 | normalized_path = path.absolute() |
| 56 | |
| 57 | locking_pids: set[int] = set() |
| 58 | path_str = str(normalized_path).lower() |
| 59 | start_time = time.time() |
| 60 | processes_checked = 0 |
| 61 | |
| 62 | # Priority process names (check these first) |
| 63 | priority_names = {"python.exe", "python3.exe", "python", "uv.exe", "uv"} |
| 64 | |
| 65 | # Get all processes and sort by priority |
| 66 | all_procs = list(psutil.process_iter(["pid", "name"])) |
| 67 | |
| 68 | # Sort processes: priority names first, then others |
| 69 | def sort_key(proc: Any) -> int: |
| 70 | try: |
| 71 | name = proc.info["name"].lower() if proc.info.get("name") else "" |
| 72 | return 0 if name in priority_names else 1 |
| 73 | except KeyboardInterrupt as ki: |
| 74 | handle_keyboard_interrupt(ki) |
| 75 | raise |
| 76 | except Exception: |
| 77 | return 2 |
| 78 | |
| 79 | all_procs.sort(key=sort_key) |
| 80 | |
| 81 | # Iterate through processes with limits |
| 82 | for proc in all_procs: |
| 83 | # Check timeout |
| 84 | if time.time() - start_time > timeout: |
| 85 | print( |
| 86 | f" Warning: Lock detection timed out after {timeout}s (checked {processes_checked} processes)" |
| 87 | ) |
no test coverage detected