Module

pgtransaction.atomic

atomic(using: _C) -> _C

atomic(
    using: str | None = None,
    savepoint: bool = True,
    durable: bool = False,
    *,
    isolation_level: Literal["SERIALIZABLE"] = ...,
    retry: int | None = None,
    read_mode: Literal["READ ONLY"] | None = None,
    deferrable: Literal["DEFERRABLE"] | None = None
) -> Atomic

atomic(
    using: str | None = None,
    savepoint: bool = True,
    durable: bool = False,
    *,
    isolation_level: (
        Literal["READ COMMITTED", "REPEATABLE READ", "SERIALIZABLE"] | None
    ) = None,
    retry: int | None = None,
    read_mode: Literal["READ WRITE", "READ ONLY"] | None = None,
    deferrable: Literal["DEFERRABLE", "NOT DEFERRABLE"] | None = None
) -> Atomic

atomic(
    using: str | None | _C = None,
    savepoint: bool = True,
    durable: bool = False,
    *,
    isolation_level: (
        Literal["READ COMMITTED", "REPEATABLE READ", "SERIALIZABLE"] | None
    ) = None,
    retry: int | None = None,
    read_mode: Literal["READ WRITE", "READ ONLY"] | None = None,
    deferrable: Literal["DEFERRABLE", "NOT DEFERRABLE"] | None = None
) -> Atomic | _C

Extends django.db.transaction.atomic with PostgreSQL functionality.

Allows one to dynamically set transaction characteristics when opening a transaction, including isolation level, read mode, and deferrability. Also supports specifying a retry policy for when an operation in that transaction results in a Postgres locking exception.

Parameters:

Name	Type	Description	Default
using	str | None | _C	The database to use.	None
savepoint	bool	If True, create a savepoint to roll back.	True
durable	bool	If True, raise a RuntimeError if nested within another atomic block.	False
isolation_level	Literal['READ COMMITTED', 'REPEATABLE READ', 'SERIALIZABLE'] | None	The isolation level we wish to be used for the duration of the transaction. If passed in as None, the current isolation level is used. Otherwise, we must choose from pgtransaction.READ_COMMITTED, pgtransaction.REPEATABLE_READ or pgtransaction.SERIALIZABLE. Note that the default isolation for a Django project is "READ COMMITTED". It is not permitted to pass this value as anything but None when using pgtransaction.atomic is used as a nested atomic block - in that scenario, the isolation level is inherited from the parent transaction.	None
retry	int | None	An integer specifying the number of attempts we want to retry the entire transaction upon encountering the settings-specified psycogp2 exceptions. If passed in as None, we default to using the settings-specified retry policy defined by settings.PGTRANSACTION_RETRY_EXCEPTIONS and settings.PGTRANSACTION_RETRY. Note that it is not possible to specify a non-zero value of retry when pgtransaction.atomic is used in a nested atomic block or when used as a context manager.	None
read_mode	Literal['READ WRITE', 'READ ONLY'] | None	The read mode for the transaction. Must be one of pgtransaction.READ_WRITE or pgtransaction.READ_ONLY.	None
deferrable	Literal['DEFERRABLE', 'NOT DEFERRABLE'] | None	Whether the transaction is deferrable. Must be one of pgtransaction.DEFERRABLE or pgtransaction.NOT_DEFERRABLE. DEFERRABLE only has effect when used with SERIALIZABLE isolation level and READ ONLY mode. In this case, it allows the transaction to be deferred until it can be executed without causing serialization anomalies.	None

Example

Since pgtransaction.atomic inherits from django.db.transaction.atomic, it can be used in exactly the same manner. Additionally, when used as a context manager or a decorator, one can use it to specify transaction characteristics. For example:

import pgtransaction

with pgtransaction.atomic(isolation_level=pgtransaction.REPEATABLE_READ):
    # Transaction is now REPEATABLE READ for the duration of the block
    ...

# Use READ ONLY mode with SERIALIZABLE isolation
with pgtransaction.atomic(
    isolation_level=pgtransaction.SERIALIZABLE,
    read_mode=pgtransaction.READ_ONLY
):
    # Transaction is now SERIALIZABLE and READ ONLY
    ...

# Use DEFERRABLE with SERIALIZABLE and READ ONLY
with pgtransaction.atomic(
    isolation_level=pgtransaction.SERIALIZABLE,
    read_mode=pgtransaction.READ_ONLY,
    deferrable=pgtransaction.DEFERRABLE
):
    # Transaction is now SERIALIZABLE, READ ONLY, and DEFERRABLE
    ...

Note that setting transaction modes in a nested atomic block is permitted as long as no queries have been made.

Example

When used as a decorator, one can also specify a retry argument. This defines the number of times the transaction will be retried upon encountering the exceptions referenced by settings.PGTRANSACTION_RETRY_EXCEPTIONS, which defaults to (psycopg.errors.SerializationFailure, psycopg.errors.DeadlockDetected). For example:

@pgtransaction.atomic(retry=3)
def update():
    # will retry update function up to 3 times
    # whenever any exception in settings.PGTRANSACTION_RETRY_EXCEPTIONS
    # is encountered. Each retry will open a new transaction (after
    # rollback the previous one).

Attempting to set a non-zero value for retry when using pgtransaction.atomic as a context manager will result in a RuntimeError.

Source code in pgtransaction/transaction.py

def atomic(
    using: str | None | _C = None,
    savepoint: bool = True,
    durable: bool = False,
    *,
    isolation_level: Literal["READ COMMITTED", "REPEATABLE READ", "SERIALIZABLE"] | None = None,
    retry: int | None = None,
    read_mode: Literal["READ WRITE", "READ ONLY"] | None = None,
    deferrable: Literal["DEFERRABLE", "NOT DEFERRABLE"] | None = None,
) -> Atomic | _C:
    """
    Extends `django.db.transaction.atomic` with PostgreSQL functionality.

    Allows one to dynamically set transaction characteristics when opening a transaction,
    including isolation level, read mode, and deferrability. Also supports specifying
    a retry policy for when an operation in that transaction results in a Postgres
    locking exception.

    Args:
        using: The database to use.
        savepoint: If `True`, create a savepoint to roll back.
        durable: If `True`, raise a `RuntimeError` if nested within another atomic block.
        isolation_level: The isolation level we wish to be
            used for the duration of the transaction. If passed in
            as None, the current isolation level is used. Otherwise,
            we must choose from `pgtransaction.READ_COMMITTED`,
            `pgtransaction.REPEATABLE_READ` or `pgtransaction.SERIALIZABLE`.
            Note that the default isolation for a Django project is
            "READ COMMITTED". It is not permitted to pass this value
            as anything but None when using [pgtransaction.atomic][]
            is used as a nested atomic block - in that scenario,
            the isolation level is inherited from the parent transaction.
        retry: An integer specifying the number of attempts
            we want to retry the entire transaction upon encountering
            the settings-specified psycogp2 exceptions. If passed in as
            None, we default to using the settings-specified retry
            policy defined by `settings.PGTRANSACTION_RETRY_EXCEPTIONS` and
            `settings.PGTRANSACTION_RETRY`. Note that it is not possible
            to specify a non-zero value of retry when [pgtransaction.atomic][]
            is used in a nested atomic block or when used as a context manager.
        read_mode: The read mode for the transaction. Must be one of
            `pgtransaction.READ_WRITE` or `pgtransaction.READ_ONLY`.
        deferrable: Whether the transaction is deferrable. Must be one of
            `pgtransaction.DEFERRABLE` or `pgtransaction.NOT_DEFERRABLE`.
            DEFERRABLE only has effect when used with SERIALIZABLE isolation level
            and READ ONLY mode. In this case, it allows the transaction to be
            deferred until it can be executed without causing serialization
            anomalies.

    Example:
        Since [pgtransaction.atomic][] inherits from `django.db.transaction.atomic`, it
        can be used in exactly the same manner. Additionally, when used as a
        context manager or a decorator, one can use it to specify transaction
        characteristics. For example:

            import pgtransaction

            with pgtransaction.atomic(isolation_level=pgtransaction.REPEATABLE_READ):
                # Transaction is now REPEATABLE READ for the duration of the block
                ...

            # Use READ ONLY mode with SERIALIZABLE isolation
            with pgtransaction.atomic(
                isolation_level=pgtransaction.SERIALIZABLE,
                read_mode=pgtransaction.READ_ONLY
            ):
                # Transaction is now SERIALIZABLE and READ ONLY
                ...

            # Use DEFERRABLE with SERIALIZABLE and READ ONLY
            with pgtransaction.atomic(
                isolation_level=pgtransaction.SERIALIZABLE,
                read_mode=pgtransaction.READ_ONLY,
                deferrable=pgtransaction.DEFERRABLE
            ):
                # Transaction is now SERIALIZABLE, READ ONLY, and DEFERRABLE
                ...

        Note that setting transaction modes in a nested atomic block is permitted as long
        as no queries have been made.

    Example:
        When used as a decorator, one can also specify a `retry` argument. This
        defines the number of times the transaction will be retried upon encountering
        the exceptions referenced by `settings.PGTRANSACTION_RETRY_EXCEPTIONS`,
        which defaults to
        `(psycopg.errors.SerializationFailure, psycopg.errors.DeadlockDetected)`.
        For example:

            @pgtransaction.atomic(retry=3)
            def update():
                # will retry update function up to 3 times
                # whenever any exception in settings.PGTRANSACTION_RETRY_EXCEPTIONS
                # is encountered. Each retry will open a new transaction (after
                # rollback the previous one).

        Attempting to set a non-zero value for `retry` when using [pgtransaction.atomic][]
        as a context manager will result in a `RuntimeError`.
    """
    # Copies structure of django.db.transaction.atomic
    if callable(using):
        return Atomic(
            using=DEFAULT_DB_ALIAS,
            savepoint=savepoint,
            durable=durable,
            isolation_level=isolation_level,
            retry=retry,
            read_mode=read_mode,
            deferrable=deferrable,
        )(using)
    else:
        return Atomic(
            using=using,
            savepoint=savepoint,
            durable=durable,
            isolation_level=isolation_level,
            retry=retry,
            read_mode=read_mode,
            deferrable=deferrable,
        )