Транзакции в Django ORM | Курс Django ORM урок 6.1

Цель урока

Разобрать механизм транзакций в Django: atomic(), savepoints, on_commit(). Понять как Django управляет транзакциями по умолчанию, когда и как явно контролировать границы транзакции, и почему on_commit() нужен для побочных эффектов, которые не должны выполняться при откате транзакции.

Необходимые знания

Урок 2.1: создание объектов
Урок 2.5: update, bulk операции
Базовое понимание ACID свойств транзакций

Как Django управляет транзакциями

По умолчанию Django работает в режиме autocommit: каждый SQL запрос автоматически оборачивается в транзакцию и фиксируется.

Product.objects.create(name="Test", price=Decimal("100"))
# Django выполняет:
# BEGIN;
# INSERT INTO shop_product (...) VALUES (...);
# COMMIT;
# - автоматически

Это поведение PostgreSQL по умолчанию (AUTOCOMMIT = True в настройках Django).

Каждый save(), update(), delete() это отдельная транзакция. Если между операциями что-то упадет, часть данных может быть сохранена, часть нет.

transaction.atomic() - явная транзакция

atomic() гарантирует что все операции внутри блока выполняются как единое целое, либо все успешно, либо откат.

from django.db import transaction

def place_order(user, cart_items):
    with transaction.atomic():
        # Создать заказ
        order = Order.objects.create(
            user=user,
            status="pending",
            total_price=sum(item.price * item.quantity for item in cart_items),
        )

        # Создать позиции заказа
        order_items = [
            OrderItem(
                order=order,
                product=item.product,
                quantity=item.quantity,
                price=item.product.price,
            )
            for item in cart_items
        ]
        OrderItem.objects.bulk_create(order_items)

        # Списать со склада
        for item in cart_items:
            Product.objects.filter(id=item.product.id).update(
                stock=F("stock") - item.quantity
            )

BEGIN;

INSERT INTO shop_order (user_id, status, total_price, ...)
VALUES (1, 'pending', 599.90, ...);

INSERT INTO shop_orderitem (order_id, product_id, quantity, price)
VALUES (1, 5, 2, 299.95), (1, 8, 1, 0.00);

UPDATE shop_product SET stock = stock - 2 WHERE id = 5;
UPDATE shop_product SET stock = stock - 1 WHERE id = 8;

COMMIT;

Если любая операция упадет с исключением, PostgreSQL выполнит ROLLBACK, и ни заказ, ни позиции, ни изменение stock не сохранятся.

atomic() как декоратор

@transaction.atomic
def transfer_stock(from_product_id, to_product_id, quantity):
    Product.objects.filter(id=from_product_id).update(
        stock=F("stock") - quantity
    )
    Product.objects.filter(id=to_product_id).update(
        stock=F("stock") + quantity
    )

Декоратор и контекстный менеджер, одно и то же. Декоратор удобнее для функций целиком, контекстный менеджер для части функции.

Savepoints - вложенные транзакции

atomic() можно вкладывать. Внешний atomic() создает транзакцию, внутренние создают savepoints:

def process_order_items(order, cart_items):
    failed_items = []

    with transaction.atomic():  # внешняя транзакция
        for item in cart_items:
            try:
                with transaction.atomic():  # savepoint
                    # Попытаться создать позицию
                    OrderItem.objects.create(
                        order=order,
                        product=item.product,
                        quantity=item.quantity,
                        price=item.product.price,
                    )
                    # Списать со склада
                    updated = Product.objects.filter(
                        id=item.product.id,
                        stock__gte=item.quantity,
                    ).update(stock=F("stock") - item.quantity)

                    if updated == 0:
                        raise ValueError(f"Недостаточно товара: {item.product.name}")

            except ValueError as e:
                # Внутренний atomic откатится до savepoint
                # Внешняя транзакция продолжается
                failed_items.append({"item": item, "error": str(e)})

    return failed_items

BEGIN;

SAVEPOINT s1;
INSERT INTO shop_orderitem (...) VALUES (...);
UPDATE shop_product SET stock = stock - 2 WHERE id = 5 AND stock >= 2;
RELEASE SAVEPOINT s1;

SAVEPOINT s2;
INSERT INTO shop_orderitem (...) VALUES (...);
UPDATE shop_product SET stock = stock - 10 WHERE id = 8 AND stock >= 10;
-- Если обновилось 0 строк - ValueError:
ROLLBACK TO SAVEPOINT s2;  -- откат только этого savepoint

COMMIT;  -- заказ и первая позиция сохранены, вторая нет

Savepoints позволяют откатить часть работы, сохранив остальное. Это полезно для обработки элементов списка, когда ошибка в одном не должна отменять другие.

Исключения внутри atomic()

Когда внутри atomic() бросается исключение, транзакция помечается как "грязная" и не может быть использована:

with transaction.atomic():
    Order.objects.create(user=user, ...)
    raise ValueError("что-то пошло не так")
    # Этот код не выполнится:
    OrderItem.objects.create(...)

После исключения Django выполняет ROLLBACK. Попытка выполнить запрос в уже "грязной" транзакции вызовет django.db.transaction.TransactionManagementError.

# Неправильно, попытка продолжить после исключения
with transaction.atomic():
    try:
        Order.objects.create(user=None, ...)  # IntegrityError
    except IntegrityError:
        pass  # поймали, думаем что всё OK

    # Следующий запрос упадет с TransactionManagementError
    # транзакция уже помечена как "грязная"
    Order.objects.count()

Решение, вложенный atomic() для части которая может упасть:

with transaction.atomic():  # внешняя транзакция
    try:
        with transaction.atomic():  # savepoint
            Order.objects.create(user=None, ...)  # IntegrityError
    except IntegrityError:
        pass  # savepoint откатился, внешняя транзакция жива

    # Продолжаем нормально
    Order.objects.count()  # OK

on_commit() - код после успешного commit

Частая ошибка, выполнять побочные эффекты внутри транзакции:

def place_order(user, cart_items):
    with transaction.atomic():
        order = Order.objects.create(user=user, ...)
        OrderItem.objects.bulk_create([...])

        # Отправить email ВНУТРИ транзакции это плохо!
        send_order_confirmation_email(user.email, order)

Если после send_order_confirmation_email транзакция откатится (например, из-за ошибки в другой части кода), email уже отправлен, а заказа нет в БД.

on_commit() регистрирует функцию которая вызывается только после успешного COMMIT:

from django.db import transaction

def place_order(user, cart_items):
    with transaction.atomic():
        order = Order.objects.create(user=user, ...)
        OrderItem.objects.bulk_create([...])

        # Email отправится только если транзакция завершится успешно
        transaction.on_commit(
            lambda: send_order_confirmation_email(user.email, order.id)
        )

Если транзакция откатится: on_commit callback не будет вызван.

on_commit() с задачами Celery

В случае запуска асинхронных задач:

from django.db import transaction

def place_order(user, cart_items):
    with transaction.atomic():
        order = Order.objects.create(user=user, ...)
        OrderItem.objects.bulk_create([...])

        # Celery задача запустится только после commit
        # Иначе задача может запуститься до того как заказ
        # появится в БД (race condition)
        transaction.on_commit(
            lambda: process_order_task.delay(order.id)
        )

Без on_commit задача Celery может запуститься до завершения транзакции. Задача попытается получить Order.objects.get(id=order.id) и не найдет запись, она ещё не закоммичена.

on_commit() в тестах

В тестах Django по умолчанию оборачивает каждый тест в транзакцию и делает ROLLBACK в конце. on_commit callback'и не вызываются в таких тестах.

Для тестирования on_commit нужно использовать TestCase.captureOnCommitCallbacks() или TransactionTestCase:

from django.test import TestCase

class OrderTestCase(TestCase):
    def test_email_sent_on_order(self):
        with self.captureOnCommitCallbacks(execute=True):
            place_order(user, cart_items)

        # Проверить что email был отправлен
        self.assertEqual(len(mail.outbox), 1)

ATOMIC_REQUESTS - транзакция на весь HTTP запрос

В settings.py можно включить автоматическую транзакцию для каждого HTTP запроса:

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",
        "ATOMIC_REQUESTS": True,  # весь запрос в транзакции
        # ...
    }
}

Весь view выполняется в одной транзакции. Если view бросает исключение, все изменения откатываются.

Плюсы: не нужно думать о транзакциях в каждом view.

Минусы:

Транзакция держится открытой на всё время view, включая медленные операции (HTTP запросы к внешним сервисам)
Долгие транзакции блокируют строки и мешают другим соединениям
on_commit() callbacks срабатывают только после окончания всего view

Для API с простыми операциями ATOMIC_REQUESTS=True удобен. Для сложных view с внешними вызовами, лучше явные atomic().

Вложенность и производительность

Каждый atomic() при вложении создает savepoint:

BEGIN;
SAVEPOINT sp1;
SAVEPOINT sp2;
-- ... операции
RELEASE SAVEPOINT sp2;
RELEASE SAVEPOINT sp1;
COMMIT;

Создание savepoint имеет небольшую стоимость. При обработке 10 000 элементов в цикле с atomic() на каждый 10 000 savepoints:

# Медленно - savepoint на каждый объект
for item in items:
    with transaction.atomic():
        process(item)

# Быстрее - одна транзакция на всё
with transaction.atomic():
    for item in items:
        process(item)

# Оптимально - батчи с обработкой ошибок
with transaction.atomic():
    for i in range(0, len(items), 500):
        batch = items[i:i + 500]
        try:
            with transaction.atomic():  # savepoint на батч
                process_batch(batch)
        except Exception:
            log_failed_batch(batch)

Практическое задание

Напишите функцию transfer_funds(from_account_id, to_account_id, amount) (представьте что у нас есть модель Account с полем balance). Используйте transaction.atomic(). Что произойдет если первый UPDATE пройдет, а второй упадет?
Реализуйте place_order(user_id, items) с транзакцией: создать Order, создать OrderItem через bulk_create, списать stock через bulk_update. Добавьте on_commit callback для отправки email.
Измените place_order чтобы она не падала полностью если отдельный товар недоступен (нет в наличии). Используйте savepoints: успешные позиции должны сохраниться, неуспешные, откататься.
Напишите тест для place_order который проверяет что email отправляется после успешного оформления заказа. Используйте captureOnCommitCallbacks().
Сравните поведение с и без on_commit():

   # Вариант A: задача запускается внутри транзакции
   with transaction.atomic():
       order = Order.objects.create(...)
       send_task.delay(order.id)  # без on_commit

   # Вариант B: задача запускается после commit
   with transaction.atomic():
       order = Order.objects.create(...)
       transaction.on_commit(lambda: send_task.delay(order.id))

Объясните что происходит в каждом варианте при откате транзакции.

Возможные ошибки

Побочные эффекты внутри транзакции

with transaction.atomic():
    order = Order.objects.create(...)
    send_email(order)        # email отправлен
    update_external_api()    # внешний API обновлен
    raise SomeException()    # транзакция откатилась!
    # В БД заказа нет, но email ушел и внешний API обновлен

Всё что нельзя откатить (HTTP запросы, файлы, email), выноси в on_commit() или после блока atomic().

Долгие операции внутри транзакции

with transaction.atomic():
    order = Order.objects.create(...)
    response = requests.post("https://payment.api/charge", ...)  # 2 секунды
    order.status = "paid"
    order.save()

HTTP запрос держит транзакцию открытой 2 секунды. Если несколько запросов обновляют одни строки, блокировки. Решение: выполняй внешние запросы до или после atomic().

Поглощение исключения без savepoint

with transaction.atomic():
    try:
        problematic_operation()
    except SomeException:
        pass  # исключение поймали, но транзакция "грязная"

    next_operation()  # TransactionManagementError

Если поглощаете исключение внутри atomic(), оберните проблемный код в вложенный atomic().

Чтение uncommitted данных

with transaction.atomic():
    Product.objects.create(name="New", ...)

# Здесь данные уже видны, commit произошел при выходе из блока
Product.objects.filter(name="New").count()  # 1

Внутри atomic() данные видны только в текущей транзакции. Другие соединения увидят их только после COMMIT.

Связь со следующим уроком

В уроке 6.2 разберем select_for_update(), явные блокировки строк. Это необходимо когда нужно прочитать строку и обновить её без race condition, например, при изменении баланса или остатков на складе.

<< Урок 5.4

Урок 6.2 >>

Подписывайтесь на мой Telegram канал

Если вам нужен ментор и вы хотите научиться веб-разработке, узнать подробнее