1
- # mypy: allow-untyped-defs
2
1
import atexit
3
2
import contextlib
4
3
from enum import Enum
23
22
import sys
24
23
import types
25
24
from types import ModuleType
25
+ from typing import Any
26
26
from typing import Callable
27
27
from typing import Dict
28
28
from typing import Iterable
59
59
)
60
60
61
61
62
- def _ignore_error (exception ) :
62
+ def _ignore_error (exception : Exception ) -> bool :
63
63
return (
64
64
getattr (exception , "errno" , None ) in _IGNORED_ERRORS
65
65
or getattr (exception , "winerror" , None ) in _IGNORED_WINERRORS
@@ -71,7 +71,7 @@ def get_lock_path(path: _AnyPurePath) -> _AnyPurePath:
71
71
72
72
73
73
def on_rm_rf_error (
74
- func ,
74
+ func : Optional [ Callable [..., Any ]] ,
75
75
path : str ,
76
76
excinfo : Union [
77
77
BaseException ,
@@ -196,7 +196,7 @@ def find_suffixes(root: Path, prefix: str) -> Iterator[str]:
196
196
return extract_suffixes (find_prefixed (root , prefix ), prefix )
197
197
198
198
199
- def parse_num (maybe_num ) -> int :
199
+ def parse_num (maybe_num : str ) -> int :
200
200
"""Parse number path suffixes, returns -1 on error."""
201
201
try :
202
202
return int (maybe_num )
@@ -264,7 +264,9 @@ def create_cleanup_lock(p: Path) -> Path:
264
264
return lock_path
265
265
266
266
267
- def register_cleanup_lock_removal (lock_path : Path , register = atexit .register ):
267
+ def register_cleanup_lock_removal (
268
+ lock_path : Path , register : Any = atexit .register
269
+ ) -> Any :
268
270
"""Register a cleanup function for removing a lock, by default on atexit."""
269
271
pid = os .getpid ()
270
272
@@ -355,7 +357,7 @@ def cleanup_candidates(root: Path, prefix: str, keep: int) -> Iterator[Path]:
355
357
yield Path (entry )
356
358
357
359
358
- def cleanup_dead_symlinks (root : Path ):
360
+ def cleanup_dead_symlinks (root : Path ) -> None :
359
361
for left_dir in root .iterdir ():
360
362
if left_dir .is_symlink ():
361
363
if not left_dir .resolve ().exists ():
@@ -459,10 +461,14 @@ def parts(s: str) -> Set[str]:
459
461
return {sep .join (parts [: i + 1 ]) or sep for i in range (len (parts ))}
460
462
461
463
462
- def symlink_or_skip (src , dst , ** kwargs ):
464
+ def symlink_or_skip (
465
+ src : Union ["os.PathLike[str]" , str ],
466
+ dst : Union ["os.PathLike[str]" , str ],
467
+ ** kwargs : Any ,
468
+ ) -> None :
463
469
"""Make a symlink, or skip the test in case symlinks are not supported."""
464
470
try :
465
- os .symlink (str ( src ), str ( dst ) , ** kwargs )
471
+ os .symlink (src , dst , ** kwargs )
466
472
except OSError as e :
467
473
skip (f"symlinks not supported: { e } " )
468
474
0 commit comments