Add concurrency stress test and minor path typing

This commit is contained in:
thePR0M3TH3AN
2025-07-01 14:10:47 -04:00
parent f7339a8fc1
commit 5269030d06
3 changed files with 84 additions and 6 deletions

View File

@@ -1,14 +1,15 @@
"""File-based locking utilities using portalocker for cross-platform support."""
from contextlib import contextmanager
from typing import Generator, Optional
from typing import Generator, Optional, Union
from os import PathLike
from pathlib import Path
import portalocker
@contextmanager
def exclusive_lock(
path: Path, timeout: Optional[float] = None
path: Union[str, PathLike[str], Path], timeout: Optional[float] = None
) -> Generator[None, None, None]:
"""Context manager that locks *path* exclusively.
@@ -29,7 +30,7 @@ def exclusive_lock(
@contextmanager
def shared_lock(
path: Path, timeout: Optional[float] = None
path: Union[str, PathLike[str], Path], timeout: Optional[float] = None
) -> Generator[None, None, None]:
"""Context manager that locks *path* with a shared lock.