angee.jobs.locks

Advisory task locks for queue workers.

LockKey

@dataclass(frozen=True)
class LockKey()

Stable advisory lock key.

name

@property
def name() -> str

Return the canonical string key used by lock backends.

LockHandle

class LockHandle(Protocol)

Owned advisory lock handle.

release

def release() -> None

Release the lock if still owned.

LockBackend

class LockBackend(Protocol)

Backend capable of acquiring one advisory task lock.

try_acquire

def try_acquire(key: LockKey,
                *,
                timeout: timedelta | None = None) -> LockHandle | None

Return a handle when key is acquired, else None.

is_held

def is_held(key: LockKey) -> bool

Return whether key is currently held.

cross_process

Whether another OS process can observe this backend's locks.

A process-local backend (the SQLite floor) makes is_held blind to a worker's locks, so callers deriving liveness from a lock must not do so unless this is true.

_LocalLockHandle

@dataclass
class _LocalLockHandle()

release

def release() -> None

Release a local lock once.

LocalLockBackend

class LocalLockBackend()

In-process lock backend for tests and SQLite development.

__init__

def __init__() -> None

Create an empty local lock table.

try_acquire

def try_acquire(key: LockKey,
                *,
                timeout: timedelta | None = None) -> LockHandle | None

Acquire key when no local caller already holds it.

release

def release(key: LockKey) -> None

Release key from the local table.

is_held

def is_held(key: LockKey) -> bool

Return whether key is held locally.

_PostgresAdvisoryLockHandle

@dataclass
class _PostgresAdvisoryLockHandle()

release

def release() -> None

Release the Postgres session-level advisory lock once.

PostgresAdvisoryLockBackend

class PostgresAdvisoryLockBackend()

Postgres session-level advisory lock backend.

Session-level advisory locks are connection-scoped. Deployments using this backend must connect directly to Postgres or use session pooling; pgbouncer transaction pooling silently breaks both is_held liveness probes and the lock-liveness checks that depend on them.

__init__

def __init__(*, alias: str = DEFAULT_DB_ALIAS) -> None

Store the Django database alias used for advisory locks.

try_acquire

def try_acquire(key: LockKey,
                *,
                timeout: timedelta | None = None) -> LockHandle | None

Acquire key through pg_try_advisory_lock.

is_held

def is_held(key: LockKey) -> bool

Return whether Postgres reports key as held.

record_lock_key

def record_lock_key(model_label: str, pk: object, purpose: str) -> LockKey

Return a stable task lock key for one model record and purpose.

task_lock

@contextmanager
def task_lock(key: LockKey,
              *,
              timeout: timedelta | None = None) -> Iterator[bool]

Yield whether the configured backend acquired key for this task.

task_locks_are_cross_process

def task_locks_are_cross_process() -> bool

Return whether the active lock backend's locks are visible across processes.

A custom backend that omits the cross_process marker is treated as process-local — the safe reading for liveness derivation.

task_lock_is_held

def task_lock_is_held(key: LockKey) -> bool

Return whether the configured backend reports key as held.

get_lock_backend

def get_lock_backend() -> LockBackend

Return the configured lock backend.