Create a lock to prevent premature folder cleanup.
(p: Path)
| 228 | |
| 229 | |
| 230 | def create_cleanup_lock(p: Path) -> Path: |
| 231 | """Create a lock to prevent premature folder cleanup.""" |
| 232 | lock_path = get_lock_path(p) |
| 233 | try: |
| 234 | fd = os.open(str(lock_path), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644) |
| 235 | except FileExistsError as e: |
| 236 | raise OSError(f"cannot create lockfile in {p}") from e |
| 237 | else: |
| 238 | pid = os.getpid() |
| 239 | spid = str(pid).encode() |
| 240 | os.write(fd, spid) |
| 241 | os.close(fd) |
| 242 | if not lock_path.is_file(): |
| 243 | raise OSError("lock path got renamed after successful creation") |
| 244 | return lock_path |
| 245 | |
| 246 | |
| 247 | def register_cleanup_lock_removal(lock_path: Path, register=atexit.register): |