import os import warnings from typing import Optional from filelock import AcquireReturnProxy from filelock import FileLock as _FileLock from .common import PathOrStr class FileLock(_FileLock): """ This is just a subclass of the `FileLock` class from the `filelock` library, except that it adds an additional argument to the `__init__` method: `read_only_ok`. By default this flag is `False`, which an exception will be thrown when a lock can't be acquired due to lack of write permissions. But if this flag is set to `True`, a warning will be emitted instead of an error when the lock already exists but the lock can't be acquired because write access is blocked. """ def __init__(self, lock_file: PathOrStr, timeout=-1, read_only_ok: bool = False) -> None: super().__init__(str(lock_file), timeout=timeout) self._read_only_ok = read_only_ok def acquire( # type: ignore[override] self, timeout=None, poll_interval=0.05, **kwargs, ) -> Optional[AcquireReturnProxy]: try: return super().acquire(timeout=timeout, poll_interval=poll_interval, **kwargs) except OSError as err: # OSError could be a lot of different things, but what we're looking # for in particular are permission errors, such as: # - errno 1 - EPERM - "Operation not permitted" # - errno 13 - EACCES - "Permission denied" # - errno 30 - EROFS - "Read-only file system" if err.errno not in (1, 13, 30): raise if os.path.isfile(self.lock_file) and self._read_only_ok: warnings.warn( f"Lacking permissions required to obtain lock '{self.lock_file}'. " "Race conditions are possible if other processes are writing to the same resource.", UserWarning, ) return None else: raise