django_subatomic.db

transaction

transaction(func: None = None, *, using: str | None = None) -> contextlib._GeneratorContextManager[None]

transaction[**P, R](func: Callable[P, R], *, using: str | None = None) -> Callable[P, R]

transaction[**P, R](func: Callable[P, R] | None = None, *, using: str | None = None) -> contextlib._GeneratorContextManager[None] | Callable[P, R]

Create a database transaction.

Can be used as a decorator or a context manager.

Nested calls are not allowed because SQL does not support nested transactions. Consider this like Django's atomic(durable=True), but with added after-commit callback support in tests.

Raises:

Type	Description
RuntimeError	if we call this from inside another existing transaction.

Source code in django_subatomic/db.py

def transaction[**P, R](
    func: Callable[P, R] | None = None, *, using: str | None = None
) -> contextlib._GeneratorContextManager[None] | Callable[P, R]:
    """
    Create a database transaction.

    Can be used as a decorator or a context manager.

    Nested calls are not allowed because SQL does not support nested transactions.
    Consider this like Django's `atomic(durable=True)`, but with added after-commit callback support in tests.

    Raises:
        RuntimeError: if we call this from inside another existing transaction.
    """

    @contextlib.contextmanager
    def _transaction(*, using: str | None) -> Generator[None]:
        with (
            _execute_on_commit_callbacks_in_tests(using),
            # Note that `savepoint=False` is not required on `atomic` because
            # the `savepoint` flag is ignored when `durable` is `True`.
            django_transaction.atomic(using=using, durable=True),
        ):
            connection = django_transaction.get_connection(using=using)
            atomic_block = connection.atomic_blocks[-1]
            atomic_block._from_subatomic = True  # noqa: SLF001
            yield

    decorator = _transaction(using=using)
    if func is None:
        return decorator
    return decorator(func)

transaction_if_not_already

transaction_if_not_already(func: None = None, *, using: str | None = None) -> contextlib._GeneratorContextManager[None]

transaction_if_not_already[**P, R](func: Callable[P, R], *, using: str | None = None) -> Callable[P, R]

transaction_if_not_already[**P, R](func: Callable[P, R] | None = None, *, using: str | None = None) -> contextlib._GeneratorContextManager[None] | Callable[P, R]

Create a transaction if one isn't already open.

Can be used as a decorator or a context manager.

Use of this hints at code which lacks control over the state it's called in.

Note

This has a bit of a clunky name. This is a deliberate attempt to discourage its use. It acts as a code-smell to highlight that places which use it may need further work to achieve full control over how transactions are managed.

Suggested alternatives

• In functions which should not control transactions, use transaction_required. This ensures they are handled by the caller.

• In functions which can unambiguously control transactions, use transaction.

Source code in django_subatomic/db.py

def transaction_if_not_already[**P, R](
    func: Callable[P, R] | None = None, *, using: str | None = None
) -> contextlib._GeneratorContextManager[None] | Callable[P, R]:
    """
    Create a transaction if one isn't already open.

    Can be used as a decorator or a context manager.

    Use of this hints at code which lacks control over the state it's called in.

    Note:
        This has a bit of a clunky name. This is a deliberate attempt to
        discourage its use. It acts as a code-smell to highlight that places
        which use it may need further work to achieve full control over how
        transactions are managed.

    Tip: Suggested alternatives
        - In functions which should not control transactions,
          use [`transaction_required`][django_subatomic.db.transaction_required].
          This ensures they are handled by the caller.

        - In functions which can unambiguously control transactions,
          use [`transaction`][django_subatomic.db.transaction].
    """

    @contextlib.contextmanager
    def _transaction_if_not_already(*, using: str | None = None) -> Generator[None]:
        if in_transaction(using=using):
            yield  # If we're already in a transaction, do nothing.
            return

        # If the innermost atomic block is from a test case, `transaction` will create a SAVEPOINT.
        # This allows for a rollback when an exception propagates out of this block, and so
        # better simulates a production transaction behaviour in tests.
        with transaction(using=using):
            yield

    decorator = _transaction_if_not_already(using=using)
    if func is None:
        return decorator
    return decorator(func)

savepoint

savepoint(*, using: str | None = None) -> Generator[None]

Create a database savepoint.

Can be used as a context manager, but not as a decorator.

Must be called inside an active transaction.

Recommended usage

• You should only create a savepoint if you may roll back to it before continuing with your transaction. If your intention is to ensure that your code is committed atomically, consider using transaction_required instead.
• We believe savepoint rollback should be handled where the savepoint is created. That locality is not possible with a decorator, so this function deliberately does not work as one.

Raises:

Type	Description
_MissingRequiredTransaction	if we are not in a transaction

Source code in django_subatomic/db.py

@_utils.contextmanager
def savepoint(*, using: str | None = None) -> Generator[None]:
    """
    Create a database savepoint.

    Can be used as a context manager, but _not_ as a decorator.

    Must be called inside an active transaction.

    Tip: Recommended usage
        - You should only create a savepoint if you may roll back to it before
          continuing with your transaction. If your intention is to ensure that
          your code is committed atomically, consider using [`transaction_required`][django_subatomic.db.transaction_required]
          instead.
        - We believe savepoint rollback should be handled where the savepoint is created.
          That locality is not possible with a decorator, so this function
          deliberately does not work as one.

    Raises:
        _MissingRequiredTransaction: if we are not in a transaction
    """
    with (
        transaction_required(using=using),
        django_transaction.atomic(using=using),
    ):
        yield

transaction_required

transaction_required(func: None = None, *, using: str | None = None) -> contextlib._GeneratorContextManager[None]

transaction_required[**P, R](func: Callable[P, R], *, using: str | None = None) -> Callable[P, R]

transaction_required[**P, R](func: Callable[P, R] | None = None, *, using: str | None = None) -> contextlib._GeneratorContextManager[None] | Callable[P, R]

Make sure that code is always executed in a transaction.

Can be used as a decorator or a context manager.

We ignore test-suite transactions when checking for a transaction because we don't want to run the risk of allowing code to pass tests but fail in production.

Raises:

Type	Description
_MissingRequiredTransaction	if we are not in a transaction.

Source code in django_subatomic/db.py

def transaction_required[**P, R](
    func: Callable[P, R] | None = None, *, using: str | None = None
) -> contextlib._GeneratorContextManager[None] | Callable[P, R]:
    """
    Make sure that code is always executed in a transaction.

    Can be used as a decorator or a context manager.

    We ignore test-suite transactions when checking for a transaction
    because we don't want to run the risk of allowing code to pass tests
    but fail in production.

    Raises:
        _MissingRequiredTransaction: if we are not in a transaction.
    """

    @contextlib.contextmanager
    def _transaction_required(*, using: str | None = None) -> Generator[None]:
        if using is None:
            using = django_db.DEFAULT_DB_ALIAS

        if not in_transaction(using=using):
            raise _MissingRequiredTransaction(database=using)
        yield

    decorator = _transaction_required(using=using)
    if func is None:
        return decorator
    return decorator(func)

durable

durable[**P, R](func: Callable[P, R]) -> Callable[P, R]

Enforce durability with this decorator.

Can be used as a decorator, but not as a context manager.

"Durability" means that the function's work cannot be rolled back after it completes, and is not to be confused with "atomicity" (which is about ensuring that the function either completes all its work or none of it).

We enforce this by ensuring that the function is not called within a transaction, and that no transaction is left open when the function completes.

Raises:

Type	Description
_UnexpectedOpenTransaction	if a transaction is already open when this is called.
_UnexpectedDanglingTransaction	if a transaction remains open after the decorated function exits. Before raising this exeption, we roll back and end the transaction.

Source code in django_subatomic/db.py

def durable[**P, R](func: Callable[P, R]) -> Callable[P, R]:
    """
    Enforce durability with this decorator.

    Can be used as a decorator, but _not_ as a context manager.

    "Durability" means that the function's work cannot be rolled back after it completes,
    and is not to be confused with "atomicity" (which is about ensuring that the function
    either completes all its work or none of it).

    We enforce this by ensuring that the function is not called within a transaction,
    and that no transaction is left open when the function completes.

    Raises:
        _UnexpectedOpenTransaction: if a transaction is already open when this is called.
        _UnexpectedDanglingTransaction: if a transaction remains open after the decorated
            function exits. Before raising this exeption, we roll back and end the
            transaction.
    """

    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        if open_dbs := dbs_with_open_transactions():
            raise _UnexpectedOpenTransaction(open_dbs=open_dbs)

        try:
            return func(*args, **kwargs)
        finally:
            if open_dbs := dbs_with_open_transactions():
                # Clean up first, otherwise we may see errors later that will mask this one.
                # This can only happen if the function manually opens a transaction,
                # so we need to manually roll it back and close it.
                for db_alias in open_dbs:
                    django_transaction.rollback(using=db_alias)
                    django_transaction.set_autocommit(True, using=db_alias)
                raise _UnexpectedDanglingTransaction(open_dbs=open_dbs)

    return wrapper

run_after_commit

run_after_commit(callback: Callable[[], object], *, using: str | None = None) -> None

Register a callback to be called after the current transaction is committed.

(Transactions created by the test suite are deliberately ignored.)

If the current transaction is rolled back, the callback will not be called.

By default, an error will be raised if there is no transaction open. While you are transitioning your codebase to stricter transaction handling, you may disable this with settings.SUBATOMIC_AFTER_COMMIT_NEEDS_TRANSACTION.

Note

Django's on_commit has a robust parameter, which allows a callback to fail silently. Kraken has a convention to "not allow code to fail silently" so this behaviour is not available from this function.

Source code in django_subatomic/db.py

def run_after_commit(
    callback: Callable[[], object],
    *,
    using: str | None = None,
) -> None:
    """
    Register a callback to be called after the current transaction is committed.

    (Transactions created by the test suite are deliberately ignored.)

    If the current transaction is rolled back, the callback will not be called.

    By default, an error will be raised if there is no transaction open.
    While you are transitioning your codebase to stricter transaction handling,
    you may disable this with [`settings.SUBATOMIC_AFTER_COMMIT_NEEDS_TRANSACTION`][subatomic_after_commit_needs_transaction].

    Note:
        Django's `on_commit` has a `robust` parameter, which allows a callback
        to fail silently. Kraken has a convention to "not allow code to fail
        silently" so this behaviour is not available from this function.
    """
    if using is None:
        using = django_db.DEFAULT_DB_ALIAS

    _ensure_transaction_is_open(using=using)

    only_in_testcase_transaction = _innermost_atomic_block_wraps_testcase(using=using)
    if (
        # See Note [Running after-commit callbacks in tests]
        getattr(settings, "SUBATOMIC_RUN_AFTER_COMMIT_CALLBACKS_IN_TESTS", True)
        and only_in_testcase_transaction
    ):
        callback()
    else:
        django_transaction.on_commit(callback, using=using)

in_transaction

in_transaction(*, using: str | None = None) -> bool

Return True if the database connection has a transaction active.

A transaction is active if the connection is no longer in autocommit mode.

So that code doesn't need to handle how testcase transactions work, testcase transactions are not considered a transaction.

Source code in django_subatomic/db.py

def in_transaction(*, using: str | None = None) -> bool:
    """
    Return `True` if the database connection has a transaction active.

    A transaction is active if the connection is no longer in autocommit mode.

    So that code doesn't need to handle how testcase transactions work,
    testcase transactions are not considered a transaction.
    """
    if using is None:
        using = django_db.DEFAULT_DB_ALIAS

    connection = django_db.connections[using]
    if connection.connection is None:
        # If there is no database connection, we can't be in a transaction.
        # We need to check this before checking for an open transaction,
        # because `get_autocommit()` would open a database connection
        # which we might not use and would consume resources unnecessarily.
        return False

    in_transaction = not django_transaction.get_autocommit(using=using)
    if not in_transaction:
        return False

    only_in_testcase_transaction = _innermost_atomic_block_wraps_testcase(using=using)

    # To make this as clear as possible I've spelled out the boolean logic here,
    # and have told ruff to ignore that this could have been simply:
    #
    #     return not only_in_testcase_transaction
    if only_in_testcase_transaction:  # noqa: SIM103
        return False
    else:
        return True

dbs_with_open_transactions

dbs_with_open_transactions() -> frozenset[str]

Get the names of databases with open transactions.

Source code in django_subatomic/db.py

def dbs_with_open_transactions() -> frozenset[str]:
    """
    Get the names of databases with open transactions.
    """
    dbs_with_open_transaction = set()
    # Note: django_db.connections is a special class which implements __iter__,
    # and should not be confused with a list or dict.
    for db_alias in django_db.connections:
        if in_transaction(using=db_alias):
            dbs_with_open_transaction.add(db_alias)

    return frozenset(dbs_with_open_transaction)