Какой декоратор нужно использовать для создания транзакции?

# Декораторы для создания транзакций в Python

Управление транзакциями — ключевой аспект работы с базами данных. Python предоставляет несколько подходов в зависимости от используемой библиотеки и ORM.

SQLAlchemy (ORM)

Декоратор с Session

Наиболее распространённый вариант в Django и SQLAlchemy проектах:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from functools import wraps

def transaction(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        session = Session()
        try:
            result = func(session, *args, **kwargs)
            session.commit()
            return result
        except Exception as e:
            session.rollback()
            raise
        finally:
            session.close()
    return wrapper

@transaction
def create_user(session, name, email):
    user = User(name=name, email=email)
    session.add(user)
    return user

Django ORM

@transaction.atomic()

Джанго предоставляет встроенный декоратор для атомарных операций:

from django.db import transaction

@transaction.atomic
def transfer_money(from_account, to_account, amount):
    from_account.balance -= amount
    from_account.save()
    
    to_account.balance += amount
    to_account.save()
    # Если возникнет ошибка после первого save — откатится

Контекстный менеджер

Альтернатива декоратору — явное управление:

from django.db import transaction

def transfer_money(from_account, to_account, amount):
    with transaction.atomic():
        from_account.balance -= amount
        from_account.save()
        
        to_account.balance += amount
        to_account.save()

asyncio + асинхронные базы данных

Для asyncpg и asyncio

import asyncpg
from functools import wraps

def async_transaction(func):
    @wraps(func)
    async def wrapper(*args, pool=None, **kwargs):
        async with pool.acquire() as conn:
            async with conn.transaction():
                return await func(conn, *args, **kwargs)
    return wrapper

@async_transaction
async def create_order(conn, user_id, items):
    order_id = await conn.fetchval(
        "INSERT INTO orders (user_id) VALUES ($1) RETURNING id",
        user_id
    )
    
    for item in items:
        await conn.execute(
            "INSERT INTO order_items (order_id, item_id, qty) VALUES ($1, $2, $3)",
            order_id, item['id'], item['qty']
        )
    
    return order_id

Psycopg3 (PostgreSQL)

С контекстным менеджером

import psycopg

def transaction_decorator(func):
    def wrapper(*args, **kwargs):
        with psycopg.connect("postgresql://...") as conn:
            with conn.transaction():
                return func(conn, *args, **kwargs)
    return wrapper

@transaction_decorator
def update_inventory(conn, product_id, quantity):
    conn.execute(
        "UPDATE products SET stock = stock - %s WHERE id = %s",
        (quantity, product_id)
    )

SQLite

import sqlite3
from contextlib import contextmanager

class Database:
    def __init__(self, db_path):
        self.db_path = db_path
    
    @contextmanager
    def transaction(self):
        conn = sqlite3.connect(self.db_path)
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()

# Использование
db = Database("app.db")

def create_user(name, email):
    with db.transaction() as conn:
        cursor = conn.cursor()
        cursor.execute(
            "INSERT INTO users (name, email) VALUES (?, ?)",
            (name, email)
        )

Обработка ошибок в транзакциях

from django.db import transaction
from django.db.utils import IntegrityError

@transaction.atomic
def create_account(user_id, initial_balance):
    try:
        account = Account.objects.create(
            user_id=user_id,
            balance=initial_balance
        )
        return account
    except IntegrityError:
        # Транзакция откатится автоматически
        raise ValueError("Account already exists")

Уровни изоляции

# Django
@transaction.atomic(durable=True)  # Durability mode
def critical_operation():
    pass

# SQLAlchemy с явной изоляцией
from sqlalchemy import event
from sqlalchemy.orm import Session

@event.listens_for(Session, "before_transaction_begin")
def receive_before_transaction_begin(session, transaction, connection):
    connection.execution_options(
        isolation_level="SERIALIZABLE"
    )

Savepoints (точки восстановления)

@transaction.atomic
def complex_operation():
    # Основная транзакция
    obj = Model.objects.create(name="test")
    
    with transaction.atomic():  # Savepoint
        try:
            risky_operation(obj)
        except Exception:
            # Откатится только внутренняя транзакция
            pass
    
    return obj

Лучшие практики

1. Минимизируй область транзакции — только необходимые операции
2. Избегай вложенных транзакций без savepoints
3. Обрабатывай специфичные исключения (IntegrityError, OperationalError)
4. Используй async транзакции для асинхронного кода
5. Избегай долгих операций внутри транзакций

Выбор декоратора зависит от ORM: Django использует @transaction.atomic(), SQLAlchemy требует кастомного декоратора с явным управлением session, а async код обычно использует встроенные методы типа conn.transaction().