meringue.core.db

PgAdvisoryLock

PgAdvisoryLock(
    lock_id: int | str,
    namespace_id: int | str | None = None,
    using: str = "default",
)

PostgreSQL advisory lock for Django projects.

This utility provides a context manager and decorator for safely acquiring PostgreSQL advisory locks based on either a single integer key or a (namespace, key) pair.

supports both signatures

• pg_advisory_lock(key) — if only lock_id is provided
• pg_advisory_lock(namespace, key) — if namespace_id is also provided

Warning

If the database is not PostgreSQL, the lock will not be applied. A warning will be issued, but the function or block will still execute normally.

Warning

If you pass strings for lock_id or namespace_id, they will be hashed to 32-bit signed ints using CRC32. Locks persist until manually released or until the DB session ends.

Examples:

As context manager

with PgAdvisoryLock(12345, namespace_id="transaction"):
    ...

As decorator

@PgAdvisoryLock(12345)
def process_transfer():
    ...

Classic

adv_lock = PgAdvisoryLock(12345)
adv_lock.lock()
...
adv_lock.unlock()

Parameters:

• lock_id (int | str) –

  Unique identifier for the resource to be locked. Strings are hashed to a 32-bit signed integer.

• namespace_id (int | str | None, default: None ) –

  Optional namespace (e.g. table name or resource type) to avoid key collisions. Also hashed if passed as a string.

• using (str, default: 'default' ) –

  Django database alias.

Source code in meringue/core/db.py

def __init__(
    self,
    lock_id: int | str,
    namespace_id: int | str | None = None,
    using: str = "default",
):
    """
    Args:
        lock_id: Unique identifier for the resource to be locked.
            Strings are hashed to a 32-bit signed integer.
        namespace_id: Optional namespace (e.g. table name or resource type)
            to avoid key collisions. Also hashed if passed as a string.
        using: Django database alias.
    """

    self.lock_id = lock_id
    self.namespace_id = namespace_id
    self.db_name = using
    self.is_postgresql = self._check_postgresql()
    self.__stacklevel = 2

int_lock_id cached property

int_lock_id: int

Returns the integer form of lock_id, applying CRC32 hash if it's a string.

Returns:

• int –

  32-bit signed integer key for pg_advisory_lock.

int_namespace_id cached property

int_namespace_id: int | None

Returns the integer form of namespace_id, applying CRC32 hash if it's a string.

Returns:

• int | None –

  Signed 32-bit integer or None if namespace is not set.

lock

lock() -> None

Acquires the advisory lock.

Source code in meringue/core/db.py

def lock(self) -> None:
    """
    Acquires the advisory lock.
    """

    if not self.is_postgresql:
        msg = "pg_advisory_lock is only supported with PostgreSQL databases."
        warnings.warn(msg, UserWarning, stacklevel=self.__stacklevel)
        return

    if self.namespace_id is None:
        sql = "SELECT pg_advisory_lock(%s)"
        args = [self.int_lock_id]
    else:
        sql = "SELECT pg_advisory_lock(%s, %s)"
        args = [self.int_namespace_id, self.int_lock_id]

    with connections[self.db_name].cursor() as cursor:
        cursor.execute(sql, args)

unlock

unlock() -> None

Releases the advisory lock.

Source code in meringue/core/db.py

def unlock(self) -> None:
    """
    Releases the advisory lock.
    """

    if not self.is_postgresql:
        return

    if self.namespace_id is None:
        sql = "SELECT pg_advisory_unlock(%s)"
        args = [self.int_lock_id]
    else:
        sql = "SELECT pg_advisory_unlock(%s, %s)"
        args = [self.int_namespace_id, self.int_lock_id]

    with connections[self.db_name].cursor() as cursor:
        cursor.execute(sql, args)

---