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)

pg_advisory_xact_lock

pg_advisory_xact_lock(
    lock_id: int | str,
    namespace_id: int | str | None = None,
    using: str = "default",
) -> None

PostgreSQL transaction-level advisory lock for Django projects.

This function uses the same key handling as PgAdvisoryLock, but calls PostgreSQL's pg_advisory_xact_lock functions. Transaction-level advisory locks are released automatically at the end of the current transaction and cannot be released manually.

supports both signatures

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

Warning

In Django autocommit mode this lock is released at the end of the SQL statement. Wrap usage in transaction.atomic() when the lock must be held for a block of application code.

Examples:

from django.db import transaction

with transaction.atomic():
    pg_advisory_xact_lock(12345, namespace_id="transfer")
    ...

Source code in meringue/core/db.py

def pg_advisory_xact_lock(
    lock_id: int | str,
    namespace_id: int | str | None = None,
    using: str = "default",
) -> None:
    """
    PostgreSQL transaction-level advisory lock for Django projects.

    This function uses the same key handling as [PgAdvisoryLock][meringue.core.db.PgAdvisoryLock],
    but calls PostgreSQL's `pg_advisory_xact_lock` functions. Transaction-level advisory locks are
    released automatically at the end of the current transaction and cannot be released manually.

    !!! info "supports both signatures"
        - `pg_advisory_xact_lock(key)`            — if only `lock_id` is provided
        - `pg_advisory_xact_lock(namespace, key)` — if `namespace_id` is also provided

    !!! warning
        In Django autocommit mode this lock is released at the end of the SQL statement. Wrap usage
        in `transaction.atomic()` when the lock must be held for a block of application code.

    Examples:
        ```py
        from django.db import transaction

        with transaction.atomic():
            pg_advisory_xact_lock(12345, namespace_id="transfer")
            ...
        ```
    """

    if not _check_postgresql(using):
        msg = "pg_advisory_xact_lock is only supported with PostgreSQL databases."
        warnings.warn(msg, UserWarning, stacklevel=2)
        return

    connection = connections[using]
    if connection.get_autocommit() and not connection.in_atomic_block:
        msg = (
            "pg_advisory_xact_lock is being used in autocommit mode. "
            "Wrap usage in transaction.atomic() to hold the lock for application code."
        )
        warnings.warn(msg, UserWarning, stacklevel=2)

    if namespace_id is None:
        sql = "SELECT pg_advisory_xact_lock(%s)"
    else:
        sql = "SELECT pg_advisory_xact_lock(%s, %s)"

    args = _advisory_lock_args(lock_id, namespace_id)

    with connection.cursor() as cursor:
        cursor.execute(sql, args)

---