FileLock

Thread-safe, platform independent file lock supporting the interface as the standard threading.Lock.

This code is loosely adapted from the py-filelock module, which is unfortunately not threadsafe.

Warning

Unlike py-filelock, this module does not provide a soft file lock to avoid any misinterpretation of the safety of using it.

Example Usage:

from aiuti.filelock import FileLock

flock = FileLock('/path/to/file.lock')

with flock:
    # Use some resource with exclusive thread and process access
class aiuti.filelock.BaseFileLock(lock_file: Union[str, PathLike], timeout: float = -1, reentrant: bool = False)

Abstract base class for a file lock.

Parameters:

  • lock_file – Path to the lock file

  • timeout – Optional timeout, in seconds, after which a Timeout exception will be raised indicating the lock couldn’t be acquired.

  • reentrant – If True, use an RLock internally to allow the same thread to acquire the lock multiple times.

property lock_file: Union[str, PathLike]

The path to the lock file.

property is_locked: bool

True, if the object holds the file lock.

acquire(blocking: bool = True, timeout: Optional[float] = None, poll_interval: float = 0.05) → bool

Acquire the file lock.

Parameters:

  • blocking – If True, block until the file lock is acquired or the timeout is exceeded. If False, return immediately if the lock cannot be acquired.

  • timeout – Optional override for timeout, the maximum number of seconds to wait for the file lock.

  • poll_interval – Number of seconds between attempts to acquire the file lock.

Returns:

True if lock acquired, False otherwise

acquire_ctx(blocking: bool = True, timeout: Optional[float] = None, poll_interval: float = 0.05) → Generator[None, None, None]

Context manager to make it more convenient to pass custom args to acquire().

Raises:

TimeoutError – If acquire() returns False.

release(force: bool = False) → None

Release the file lock.

Note

The file lock is only completely released if the lock counter is zero when using reentrant locks.

Note

The lock file itself is not automatically deleted to avoid race conditions.

Parameters:

force – If true, the lock counter is ignored and the lock is released in every case.

class aiuti.filelock.WindowsFileLock(lock_file: Union[str, PathLike], timeout: float = -1, reentrant: bool = False)

Windows specific file lock implementation which uses msvcrt.locking() to hard lock the lock file.

Parameters:

  • lock_file – Path to the lock file

  • timeout – Optional timeout, in seconds, after which a Timeout exception will be raised indicating the lock couldn’t be acquired.

  • reentrant – If True, use an RLock internally to allow the same thread to acquire the lock multiple times.

acquire(blocking: bool = True, timeout: Optional[float] = None, poll_interval: float = 0.05) → bool

Acquire the file lock.

Parameters:

  • blocking – If True, block until the file lock is acquired or the timeout is exceeded. If False, return immediately if the lock cannot be acquired.

  • timeout – Optional override for timeout, the maximum number of seconds to wait for the file lock.

  • poll_interval – Number of seconds between attempts to acquire the file lock.

Returns:

True if lock acquired, False otherwise

acquire_ctx(blocking: bool = True, timeout: Optional[float] = None, poll_interval: float = 0.05) → Generator[None, None, None]

Context manager to make it more convenient to pass custom args to acquire().

Raises:

TimeoutError – If acquire() returns False.

property is_locked: bool

True, if the object holds the file lock.

property lock_file: Union[str, PathLike]

The path to the lock file.

release(force: bool = False) → None

Release the file lock.

Note

The file lock is only completely released if the lock counter is zero when using reentrant locks.

Note

The lock file itself is not automatically deleted to avoid race conditions.

Parameters:

force – If true, the lock counter is ignored and the lock is released in every case.

class aiuti.filelock.UnixFileLock(lock_file: Union[str, PathLike], timeout: float = -1, reentrant: bool = False)

Unix specific file lock implementation which uses fcntl.flock() to hard lock the lock file.

Parameters:

  • lock_file – Path to the lock file

  • timeout – Optional timeout, in seconds, after which a Timeout exception will be raised indicating the lock couldn’t be acquired.

  • reentrant – If True, use an RLock internally to allow the same thread to acquire the lock multiple times.

acquire(blocking: bool = True, timeout: Optional[float] = None, poll_interval: float = 0.05) → bool

Acquire the file lock.

Parameters:

  • blocking – If True, block until the file lock is acquired or the timeout is exceeded. If False, return immediately if the lock cannot be acquired.

  • timeout – Optional override for timeout, the maximum number of seconds to wait for the file lock.

  • poll_interval – Number of seconds between attempts to acquire the file lock.

Returns:

True if lock acquired, False otherwise

acquire_ctx(blocking: bool = True, timeout: Optional[float] = None, poll_interval: float = 0.05) → Generator[None, None, None]

Context manager to make it more convenient to pass custom args to acquire().

Raises:

TimeoutError – If acquire() returns False.

property is_locked: bool

True, if the object holds the file lock.

property lock_file: Union[str, PathLike]

The path to the lock file.

release(force: bool = False) → None

Release the file lock.

Note

The file lock is only completely released if the lock counter is zero when using reentrant locks.

Note

The lock file itself is not automatically deleted to avoid race conditions.

Parameters:

force – If true, the lock counter is ignored and the lock is released in every case.

aiuti.filelock.FileLock

Alias for the lock, which should be used for the current platform. On Windows, this is an alias for WindowsFileLock and on Unix for UnixFileLock.