Create a numbered dir with a cleanup lock and remove old ones.
(
root: Path,
prefix: str,
keep: int,
lock_timeout: float,
mode: int,
)
| 346 | |
| 347 | |
| 348 | def make_numbered_dir_with_cleanup( |
| 349 | root: Path, |
| 350 | prefix: str, |
| 351 | keep: int, |
| 352 | lock_timeout: float, |
| 353 | mode: int, |
| 354 | ) -> Path: |
| 355 | """Create a numbered dir with a cleanup lock and remove old ones.""" |
| 356 | e = None |
| 357 | for i in range(10): |
| 358 | try: |
| 359 | p = make_numbered_dir(root, prefix, mode) |
| 360 | lock_path = create_cleanup_lock(p) |
| 361 | register_cleanup_lock_removal(lock_path) |
| 362 | except Exception as exc: |
| 363 | e = exc |
| 364 | else: |
| 365 | consider_lock_dead_if_created_before = p.stat().st_mtime - lock_timeout |
| 366 | # Register a cleanup for program exit |
| 367 | atexit.register( |
| 368 | cleanup_numbered_dir, |
| 369 | root, |
| 370 | prefix, |
| 371 | keep, |
| 372 | consider_lock_dead_if_created_before, |
| 373 | ) |
| 374 | return p |
| 375 | assert e is not None |
| 376 | raise e |
| 377 | |
| 378 | |
| 379 | def resolve_from_str(input: str, rootpath: Path) -> Path: |
no test coverage detected