Skip to content

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
  • pg_advisory_lock(key) — if only lock_id is provided
  • pg_advisory_lock(namespace, key) — if namespace_id is also provided

Example usage:

with PgAdvisoryLock("user_42", namespace_id="payment"):
    # Critical section here...
    ...

@PgAdvisoryLock(12345)
def process_transfer():

Note

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.

Note

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.

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)