Warning
You are reading the documentation for v0.0.2. The latest version is v0.0.15.
Source code for dreadlocks.path_lock
from contextlib import contextmanager
from os.path import normpath
from .thread_level_lock import thread_level_lock
from .process_level_path_lock import (
_process_level_path_lock, # type: ignore [reportPrivateUsage]
)
[docs]
@contextmanager
def path_lock(
path: str, shared: bool = False, blocking: bool = True, reentrant: bool = False
):
"""Locks a path both at the thread-level and process-level.
Parameters
----------
path
The path to lock. This path is open in write mode to ensure its
existence in an atomic way. Cannot be a path to an existing directory.
The path is normalized so that the underlying file descriptor and
thread-level lock can be keyed.
shared
Whether the lock is shared. If not shared, the lock is
exclusive. One can have either no locks, a single exclusive lock,
or any positive number of shared locks on the same resource at any given
time.
blocking
Whether lock acquisition is blocking. If True, will raise an
error if a lock cannot be acquired immediately. Otherwise, will block
until the lock is acquired.
reentrant
Whether lock acquisition is reentrant. If True, allows to lock
recursively. Otherwise, locking recursively results in a dead-lock.
"""
normalized_path = normpath(path)
with thread_level_lock(normalized_path, shared, blocking, reentrant):
with _process_level_path_lock(
normalized_path, shared, blocking, reentrant
) as fd:
yield fd