ddp_utils.runtime.lock

import ddp_utils.runtime.lock

Provide an interprocess lock based on atomic lock-file creation.

FileLock records owner metadata, detects stale locks, and supports explicit and context-managed acquisition lifecycles.

Examples

Use this public operation:

from ddp_utils.runtime.lock import FileLock

lock = FileLock("example", "update", create_now=False)

class ddp_utils.runtime.lock.FileLock(namespace: str, name: str, *, timeout: float | None = 30.0, poll_interval: float = 0.2, stale_after: float | None = None, create_now: bool = True)

Bases: object

Cross-process file lock based on atomic lock-file creation.

The lock is stored under the namespace locks root. The lock file contains metadata about the owning process. Lock acquisition is based on atomic file creation using O_CREAT | O_EXCL, which is supported by the operating system and is sufficient for many production workflows where a lightweight advisory lock is needed.

Typical use cases:

• preventing two processes from downloading the same driver simultaneously;

• serializing access to a shared cache directory;

• guarding write operations to shared runtime artifacts.

The lock file layout is plain text with simple key/value pairs.

Parameters

Name	Type	Description
namespace	str	Namespace of the owning library or module.
name	str	Logical lock name. It is normalized into a filesystem-safe file name.
timeout	Optional[float]	Maximum number of seconds to wait for acquiring the lock. If None, wait indefinitely.
poll_interval	float	Delay between acquisition attempts.
stale_after	Optional[float]	If provided, a lock file older than this number of seconds may be considered stale and removed before retrying acquisition.
create_now	bool	If True, the lock directory is created immediately.

Raises

Exception	Description
ValueError	If name is empty.

Examples

Use this public operation:

instance = FileLock(...)

Configure one namespaced lock and its acquisition policy.

The lock itself is not acquired during initialization. Resolving its path ensures the namespace lock directory exists even when create_now is false.

Parameters

Name	Type	Description
namespace	str	Non-empty runtime namespace containing the lock.
name	str	Logical lock name normalized for use as a filename.
timeout	Optional[float]	Maximum acquisition wait in seconds, or None to wait indefinitely.
poll_interval	float	Delay in seconds between acquisition attempts.
stale_after	Optional[float]	Optional age in seconds after which an existing lock may be removed as stale.
create_now	bool	Create the complete runtime directory layout eagerly. The lock directory is still created while resolving path.

Raises

Exception	Description
ValueError	namespace or name is empty or invalid.
OSError	The runtime lock directory cannot be created.

Examples

Configure a bounded lock without acquiring it:

lock = FileLock("reports", "refresh", timeout=5.0)

property path: Path

Return the lock file path.

Returns

Path to the lock file.

Examples

Use this public operation:

result = instance.path()

property acquired: bool

Return whether the current instance owns the lock.

Returns

True if the lock is currently held by this instance.

Examples

Use this public operation:

result = instance.acquired()

acquire() → bool

Acquire the lock.

If a stale lock policy is configured through stale_after, stale locks are removed before retrying acquisition.

Returns

Type	Description
bool	True if the lock was acquired.

Raises

Exception	Description
TimeoutError	If the timeout is reached before acquiring the lock.

Examples

Use this public operation:

result = instance.acquire()

release() → None

Release the lock if it is owned by this instance.

The method is safe to call multiple times.

Examples

Use this public operation:

result = instance.release()

force_release() → None

Remove the lock file unconditionally.

This should be used only in controlled scenarios when a caller knows that the current lock file is no longer valid.

Examples

Use this public operation:

result = instance.force_release()

exists() → bool

Check whether a lock file currently exists.

Returns

Type	Description
bool	True if the lock file exists.

Examples

Use this public operation:

result = instance.exists()

age_seconds() → float | None

Return the current lock age in seconds.

Returns

Type	Description
float | None	Lock age in seconds, or None if the lock does not exist.

Examples

Use this public operation:

result = instance.age_seconds()

is_stale() → bool

Check whether the lock is stale according to stale_after.

Returns

Type	Description
bool	True if the lock exists and is older than stale_after.

Examples

Use this public operation:

result = instance.is_stale()

read_metadata() → dict[str, str]

Read lock metadata from the lock file.

Returns

Type	Description
dict[str, str]	Dictionary of key/value pairs. Empty dict if file does not exist or cannot be parsed.

Examples

Use this public operation:

result = instance.read_metadata()