cdxcore.filelock#

Simple file-based system-wide lock for both Linux and Windows.

Overview#

The most effective method of using filelock is calling cdxcore.filelock.AttemptLock() in a context block:

from cdxcore.filelock import FileLock
from cdxcore.subdir import SubDir
lock_dir  = SubDir("!/locks",ext="lck")
lock_name = lock_dir.full_file_name("lock1")

with AcquireLock( lock_name, timeout_second=2, timeout_retry=3 ):
    # do locked activity

# locked section over

In above example the function cdxcore.filelock.AcquireLock() will attempt to acquire a file lock using the file lock_name in three attempts with a timeout of two seconds between them. If acquiring a lock fails, a BlockingIOError is raised.

If successful, the with construct ensures that the lock is released at the end of the block.

If we can handle a situation where a lock is not acquired safely, the following pattern using cdxcore.filelock.AttemptLock() can bs used:

from cdxcore.filelock import FileLock
from cdxcore.subdir import SubDir
lock_dir  = SubDir("!/locks",ext="lck")
lock_name = lock_dir.full_file_name("lock1")

with AttemptLock( lock_name, timeout_second=2, timeout_retry=3 ) as lock:
    if lock.acquired:
        # do locked activity
    else:
        # not locked activity

# locked section over

Multiple Acquisitions

Im both patterns above the lock was acquired only once. If the lock is to be acquired several times, or to be passed to other functions, it is better to first create a cdxcore.filelock.Flock object and then use cdxcore.filelock.FLock.acquire() instead of cdxcore.filelock.AcquireLock():

from cdxcore.filelock import FileLock
from cdxcore.subdir import SubDir

def subroutine( lock ):
    with lock.aquire( timeout_second=2, timeout_retry=3 ):
        # do locked activity

def mainroutine():
    lock_dir  = SubDir("!/locks",ext="lck")
    lock_name = lock_dir.full_file_name("lock1")
    lock      = FileLock(lock_name)

    with lock.aquire( timeout_second=2, timeout_retry=3 ):
        # do locked activity

    subroutine( lock )

In this case, cdxcore.filelock.FLock.attempt() can be used for conditional workflows based on lock status:

def subroutine( lock ):
    with lock.attempt( lock_name, timeout_second=2, timeout_retry=3 ) as lh:
        if not lh.acquired:
            return # job already done
        # do locked activity

Explicit State Management

The use of with context blocks ensures that locks are released as soon as the protected activity is finished. In some cases we may desired to finely control such workflow. In this case, use cdxcore.filelock.FLock.acquire() and cdxcore.filelock.FLock.release() in pairs:

from cdxcore.filelock import FileLock
from cdxcore.subdir import SubDir

lock_dir  = SubDir("!/locks",ext="lck")
lock_name = lock_dir.full_file_name("lock1")
lock      = FileLock(lock_name)

def subroutine( lock ):
    if not lock.acquire( timeout_second=2, timeout_retry=3 ):
        return
    # do protected work
    lock.release()

try:    
    if lock.acquire( timeout_second=2, timeout_retry=3 ):
        # do some protected work
        lock.release()

    ...

    subroutine(lock)
    ...

    if lock.acquire( timeout_second=2, timeout_retry=3 ):
        # do some protected work
        lock.release()

finally:
    lock.clear() # <- clears all acquisitions of the lock and stops further use.            

Garbage Collection

By default locks will delete the underlying file using cdxcore.filelock.FLock.clear() upon garbage collection. This can be triggered with gc.collect().

Import#

from cdxcore.filelock import FileLock, AcquireLock

Documentation#

Functions

`AcquireLock`(filename, *[, wait, ...])	Acquire a file lock and return a context handler, or raise an exception. The context handler can be used in a `with` statement as follows::.
`AttemptLock`(filename, *[, wait, ...])	Attempt to acquire a file lock and return a context handler even if the lock was not acquired.
`FileLock`(filename, *[, release_on_exit, verbose])	Acquire a file lock object shared among threads.

Classes

`FLock`(filename, *[, release_on_exit, verbose])	System-wide file lock.
`LockContext`(flock, acquired)	A context handler returned by `cdxcore.filelock.Flock.acquire()`, `cdxcore.filelock.AcquireLock()`, and `cdxcore.filelock.AttemptLock()`.

cdxcore.filelock.AcquireLock(filename, *, wait=True, timeout_seconds=None, timeout_retry=None, verbose=None)[source]#

Acquire a file lock and return a context handler, or raise an exception. The context handler can be used in a with statement as follows:

from cdxcore.filelock import FileLock
from cdxcore.subdir import SubDir

lock_dir  = SubDir("!/locks",ext="lck")
lock_name = lock_dir.full_file_name("lock1")

with AcquireLock( lock_name, timeout_second=2, timeout_retry=3 ):
    # do locked activity
# no longer locked

Note that this function will raise an exception if the lock could be acquired. Use cdxcore.filelock.AttemptLock() to obtain a context handler even if the lock was not acquired.

Parameters:

filenamestr

Filename of the lock.

filename may start with '!/' to refer to the temp directory, or '~/' to refer to the user directory. On Unix '/dev/shm/' can be used to refer to the standard shared memory directory in case a shared memory file is being locked.

waitbool, default True

If False, return immediately if the lock cannot be acquired.
If True, wait with below parameters; in particular if these are left as defaults the lock will wait indefinitely.

timeout_secondsint | None, default None

Number of seconds to wait before retrying. Set to 0` to fail immediately. If set to None, then behaviour will depend on wait:

If wait is True, then timeout_seconds==1.
If wait is False, then timeout_seconds==0.

timeout_retryint | None, default None

How many times to retry before timing out. Set to None to retry indefinitely.

verbosecdxcore.verbose.Context | None, default None

Context which will print out operating information of the lock. This is helpful for debugging. In particular, it will track __del__() function calls. Set to None to supress printing any context.

Returns:

Contextcdxcore.filelock.LockContext: A context representing the acquired state which can be used with with. The function cdxcore.filelock.release() is called at the end of the with statement to release the acquired lock.

Raises:

TimeoutTimeoutError: Raised if acquire is True, if timeout_seconds > 0 and wait==True, and if the call failed to obtain the file lock.
BlockedBlockingIOError: Raised if acquire is True, if timeout_seconds == 0 or wait==False, and if the call failed to obtain the file lock.

cdxcore.filelock.AttemptLock(filename, *, wait=True, timeout_seconds=None, timeout_retry=None, verbose=None)[source]#

Attempt to acquire a file lock and return a context handler even if the lock was not acquired. The context handler’s cdxcore.filelock.LockContext.acquired can be used to assess whether the lock was acquired.

The pattern is as follows:

from cdxcore.filelock import FileLock
from cdxcore.subdir import SubDir

lock_dir  = SubDir("!/locks",ext="lck")
lock_name = lock_dir.full_file_name("lock1")

with AttemptLock( lock_name, timeout_second=2, timeout_retry=3 ) as lock:
    if lock.acquired:
        # do locked activity
    else:
        # do not locked activity
# no longer locked

Parameters:

filenamestr

Filename of the lock.

filename may start with '!/' to refer to the temp directory, or '~/' to refer to the user directory. On Unix '/dev/shm/' can be used to refer to the standard shared memory directory in case a shared memory file is being locked.

waitbool, default True

If False, return immediately if the lock cannot be acquired.
If True, wait with below parameters; in particular if these are left as defaults the lock will wait indefinitely.

timeout_secondsint | None, default None

Number of seconds to wait before retrying. Set to 0` to fail immediately. If set to None, then behaviour will depend on wait:

If wait is True, then timeout_seconds==1.
If wait is False, then timeout_seconds==0.

timeout_retryint | None, default None

How many times to retry before timing out. Set to None to retry indefinitely.

verbosecdxcore.verbose.Context | None, default None

Context which will print out operating information of the lock. This is helpful for debugging. In particular, it will track __del__() function calls. Set to None to supress printing any context.

Returns:

Filelock if acquired or None

class cdxcore.filelock.FLock(filename, *, release_on_exit=True, verbose=None)[source]#

Bases: object

System-wide file lock.

Do not construct members of this class directly as it will not be able to create a second lock on the same lock file within the same process. Use the “factory” function cdxcore.filelock.FileLock() instead.

Attributes:

filename: Return the filename of the lock.
locked: Whether the lock is active.
num_acquisitions: Returns the net number of times the file was acquired using cdxcore.filelock.FLock.acquire().

Methods

`acquire`([wait, timeout_seconds, ...])	Acquire lock.
`attempt`([wait, timeout_seconds, timeout_retry])	Attempt to acquire lock.
`clear`()	Clears the current object and forces its release.
`release`(*[, force])	Release lock.

acquire(wait=True, *, timeout_seconds=1, timeout_retry=5, raise_on_fail=True)[source]#

Acquire lock.

If successful, this function returns a cdxcore.filelock.LockContext which can be used in a with statement as follows:

from cdxcore.filelock import FileLock
from cdxcore.subdir import SubDir

lock_dir  = SubDir("!/locks",ext="lck")
lock_name = lock_dir.full_file_name("lock1")
lock      = FileLock(lock_name)

with lock.aquire( timeout_second=2, timeout_retry=3 ):
    # do locked activity
# no longer locked

In case acquire() fails to obtain the lock, by default it will raise an exception.

One-Shot

If you only acquire a lock once, it is more convenient to use cdxcore.filelock.AcquireLock():

from cdxcore.filelock import FileLock
from cdxcore.subdir import SubDir

lock_dir  = SubDir("!/locks",ext="lck")
lock_name = lock_dir.full_file_name("lock1")

with AcquireLock( lock_name, timeout_second=2, timeout_retry=3 ):
    # do locked activity
# no longer locked

Parameters:

waitbool, default True

If False, return immediately if the lock cannot be acquired.
If True, wait with below parameters; in particular if these are left as defaults the lock will wait indefinitely.

timeout_secondsint | None, default None

Number of seconds to wait before retrying. Set to 0` to fail immediately. If set to None, then behaviour will depend on wait:

If wait is True, then timeout_seconds==1.
If wait is False, then timeout_seconds==0.

timeout_retryint | None, default None

How many times to retry before timing out. Set to None to retry indefinitely.

raise_on_failbool, default True

By default, if the constructor fails to obtain the lock, raise an exception. This will be either of type

TimeoutError if timeout_seconds > 0 and wait==True, or
BlockingIOError if timeout_seconds == 0 or wait==False.

If the function could not acquire a lock on the file and if raise_on_fail is False, then this function returns None. This can be used for manual control workflows.

Returns:

Contextcdxcore.filelock.LockContext

A context manager representing the acquired state which can be used with with. If the context manager protocol os used, then cdxcore.filelock.release() is called at the end of the with statement.

This function returns None if the lock could be acquired and raise_on_fail is False.-heut3//..X The method cdxcore.filelock.FLock.attempt() will return an unacquired context manager in case of a failure.

Raises:

TimeoutTimeoutError: Raised if acquire is True, if timeout_seconds > 0 and wait==True, and if the call failed to obtain the file lock.
BlockedBlockingIOError: Raised if acquire is True, if timeout_seconds == 0 or wait==False, and if the call failed to obtain the file lock.

attempt(wait=True, *, timeout_seconds=1, timeout_retry=5)[source]#

Attempt to acquire lock.

This function attempts to obtain the file lock within the specified timeout parameters. It will return a cdxcore.filelock.LockContext whose property cdxcore.filelock.LockContext.acquired provides success of this attempt.

The context object can be used using with as follows:

from cdxcore.filelock import FileLock from cdxcore.subdir import SubDir

lock_dir = SubDir(“!/locks”,ext=”lck”) lock_name = lock_dir.full_file_name(“lock1”) lock = FileLock(lock_name)

with lock.attempt( timeout_second=2, timeout_retry=3 ) as lh:

if lh.acquired:
# do locked activity

else:
# do some other activity; warn the user; etc

# no longer locked

In contrast, the function cdxcore.filelock.FLock.acquire() will only return a cdxcore.filelock.LockContext object if the acquisiton of the lock was successful.

One-Shot

If you only make one attempt to use a lock, it is more convenient to use cdxcore.filelock.AttemptLock():

with AttemptLock( lock_name, timeout_second=2, timeout_retry=3 ) as lock:
    if lock.acquired:
        # do locked activity
    else:
        # do not locked activity
# no longer locked

Parameters:

waitbool, default True

If False, return immediately if the lock cannot be acquired.
If True, wait with below parameters; in particular if these are left as defaults the lock will wait indefinitely.

timeout_secondsint | None, default None

Number of seconds to wait before retrying. Set to 0` to fail immediately. If set to None, then behaviour will depend on wait:

If wait is True, then timeout_seconds==1.
If wait is False, then timeout_seconds==0.

timeout_retryint | None, default None

How many times to retry before timing out. Set to None to retry indefinitely.

Returns:

Contextcdxcore.filelock.LockContext

A context representing the acquired state which can be used with with. Check cdxcore.filelock.LockContext.acquired to validate whether the lock was acquired successfully.

If with is used and cdxcore.filelock.LockContext.acquired is True, then cdxcore.filelock.release() is called at the end of the with statement to release the acquired lock.

clear()[source]#: Clears the current object and forces its release. Will delete the underlying lock file if release_on_exit was used when constructing the lock.

property filename: str#: Return the filename of the lock.

property locked: bool#: Whether the lock is active.

property num_acquisitions: int#: Returns the net number of times the file was acquired using cdxcore.filelock.FLock.acquire(). Zero if the lock is not currently held.

release(*, force=False)[source]#

Release lock.

By default this function will only decreased the number of successful acquisitions by one, and will delete the file lock only once the number of acquisitions is zero. Use force to force an unlock.

Parameters:

forcebool, default: False: Whether to close the file regardless of the internal acquisition counter.

Returns:

Remainingint: Returns numbner of remaining lock acquisitions; in other words returns 0 if the lock is no longer locked by this process.

cdxcore.filelock.FileLock(filename, *, release_on_exit=True, verbose=None)[source]#

Acquire a file lock object shared among threads.

This function is useful if a lock is going the be used iteratively, including passing it to sub-routines:

from cdxcore.filelock import FileLock
from cdxcore.subdir import SubDir

def subroutine( lock ):
    with lock.aquire( lock_name, timeout_second=2, timeout_retry=3 ):
        # do locked activity

def mainroutine():
    lock_dir  = SubDir("!/locks",ext="lck")
    lock_name = lock_dir.full_file_name("lock1")
    lock      = FileLock(lock_name)

    with lock.aquire( timeout_second=2, timeout_retry=3 ):
        # do locked activity

    subroutine( lock )

If the lock is only used for a one-of acquisition, it is usally prettier to use cdxcore.filelock.AcquireLock() instead.

Parameters:

filenamestr

Filename of the lock.

filename may start with '!/' to refer to the temp directory, or '~/' to refer to the user directory. On Unix '/dev/shm/' can be used to refer to the standard shared memory directory in case a shared memory file is being locked.

release_on_exitbool, default True

Whether to auto-release the lock upon exit.

verbosecdxcore.verbose.Context | None, default None

Context which will print out operating information of the lock. This is helpful for debugging. In particular, it will track __del__() function calls. Set to None to supress printing any context.

Returns:

lockcdxcore.filelock.FLock: The lock. This function will re-use an existing lock if it has been created elsewhere by the same process.

class cdxcore.filelock.LockContext(flock, acquired)[source]#

Bases: object

A context handler returned by cdxcore.filelock.Flock.acquire(), cdxcore.filelock.AcquireLock(), and cdxcore.filelock.AttemptLock().

Attributes:

acquired: Whether the underlying file lock was acquired by this context handler.
filename: Underlying file lock name

property acquired: bool#: Whether the underlying file lock was acquired by this context handler. This is might be False for LockContext objects returned by cdxcore.filelock.AttemptLock()

property filename: str#: Underlying file lock name

cdxcore.filelock#

Overview#

Import#

Documentation#

This Page