Как в SQLAlchemy выполнить несколько операций записи в рамках одной транзакции?

Транзакции в SQLAlchemy: основной концепт

В SQLAlchemy (версия 2.0+) транзакции управляются автоматически через session или Engine. Все операции write-операции (INSERT, UPDATE, DELETE) выполняются в рамках одной транзакции и либо коммитятся все вместе, либо откатываются при ошибке.

Способ 1: Использование Session с контекстным менеджером (рекомендуется)

from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from models import User, Post

engine = create_engine("postgresql://user:password@localhost/db")

def create_user_with_posts():
    with Session(engine) as session:
        user = User(name="Alice", email="alice@example.com")
        session.add(user)
        session.flush()
        
        post = Post(title="First Post", user_id=user.id)
        session.add(post)
        
        another_post = Post(title="Second Post", user_id=user.id)
        session.add(another_post)

Способ 2: Явный контроль через begin()

from sqlalchemy.orm import Session

def atomic_operation():
    with Session(engine) as session:
        with session.begin():
            user = User(name="Bob")
            session.add(user)
            session.flush()
            
            if user.id:
                post = Post(title="Post", user_id=user.id)
                session.add(post)

Способ 3: Управление вручную

with Session(engine) as session:
    try:
        session.add(user)
        session.add(post)
        session.commit()
    except Exception as e:
        session.rollback()
        raise

Способ 4: Асинхронный код

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

async_engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost/db"
)

async def create_records_async():
    async with AsyncSession(async_engine) as session:
        async with session.begin():
            user = User(name="Charlie")
            session.add(user)
            await session.flush()
            
            post = Post(title="Async Post", user_id=user.id)
            session.add(post)

Ключевые моменты

flush() vs commit()

• flush() отправляет SQL в БД, но не коммитит транзакцию
• commit() фиксирует все изменения

Savepoints для частичного отката

with Session(engine) as session:
    user = User(name="David")
    session.add(user)
    
    savepoint = session.begin_nested()
    
    try:
        risky_post = Post(title="Risky", user_id=user.id)
        session.add(risky_post)
    except Exception:
        savepoint.rollback()
        session.add(safe_post)
    
    session.commit()

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

1. Всегда используйте контекстный менеджер
2. Избегайте глобальных сессий
3. Вызывайте flush() для получения ID перед использованием
4. Используйте savepoints для сложной логики
5. Обрабатывайте исключения правильно

Обработка ошибок

from sqlalchemy.exc import IntegrityError

def safe_create_records():
    with Session(engine) as session:
        try:
            user = User(name="Eve", email="eve@example.com")
            session.add(user)
            session.flush()
            
            post = Post(title="Post", user_id=user.id)
            session.add(post)
            session.commit()
        except IntegrityError as e:
            session.rollback()
            print(f"Duplicate entry: {e}")
            raise
        except Exception as e:
            session.rollback()
            raise

Вывод: используйте контекстные менеджеры и полагайтесь на автоматический коммит. Это безопасный и читаемый подход.