stouputils.lock.shared module#

exception LockError[source]#

Bases: RuntimeError

Base lock error.

exception LockTimeoutError[source]#

Bases: TimeoutError, LockError

Raised when a lock could not be acquired within timeout seconds.

resolve_path(path: str) → str[source]#

Resolve a lock file path, placing it in the system temporary directory if only a name is given.

>>> import os, tempfile
>>> p = resolve_path('foo.lock')
>>> os.path.basename(p) == 'foo.lock'
True
resolve_acquire_defaults(
blocking_arg: bool | None,
timeout_arg: float | None,
check_interval_arg: float | None,
default_blocking: bool,
default_timeout: float | None,
default_check_interval: float,
) → tuple[bool, float | None, float, float | None][source]#

Resolve acquire() parameter defaults and compute the deadline.

Returns:

(blocking, timeout, check_interval, deadline)

wait_or_raise(
blocking: bool,
deadline: float | None,
check_interval: float,
name: str,
) → None[source]#

Sleep one check interval before the next attempt on a busy lock, or raise when waiting is not allowed.

Parameters:

  • deadline – time.monotonic() value past which waiting stops, None to wait forever.

  • name – Lock name or path the timeout message reports.

Raises:

LockTimeoutError – If blocking is False or deadline has passed.

>>> wait_or_raise(False, None, 0.0, "foo.lock")
Traceback (most recent call last):
        ...
stouputils.lock.shared.LockTimeoutError: Lock is already held and blocking is False
>>> wait_or_raise(True, 0.0, 0.0, "foo.lock")
Traceback (most recent call last):
        ...
stouputils.lock.shared.LockTimeoutError: Timeout while waiting for lock 'foo.lock'