Приклад роботи з транзакціями та rollback у SQLAlchemy

Давайте уявимо класичну ситуацію з життя розробника: у вас є банківський додаток, і треба перевести гроші з одного рахунку на інший. Очевидно, що ця операція складається з таких кроків:

1. Знімаємо гроші з одного рахунку.
2. Кладаємо гроші на інший рахунок.

Якщо на етапі зняття грошей все пройшло добре, а на етапі зарахування сталася помилка — це катастрофа. Гроші кудись зникли! Саме тут нам на допомогу приходять транзакції.

Для початку створимо просту модель Account, щоб почати роботу з транзакціями.

from sqlalchemy import Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Account(Base):
    __tablename__ = 'accounts'

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    balance = Column(Float, default=0.0)

Тепер у нас є таблиця акаунтів з балансом. Припустимо, треба реалізувати функцію, яка переводить гроші між двома акаунтами. Використаємо транзакцію для забезпечення атомарності операції:

from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError

def transfer_money(session: Session, from_account_id: int, to_account_id: int, amount: float):
    try:
        # Починаємо транзакцію
        from_account = session.query(Account).filter(Account.id == from_account_id).first()
        to_account = session.query(Account).filter(Account.id == to_account_id).first()

        if from_account is None or to_account is None:
            raise ValueError("Один з акаунтів не знайдено!")

        if from_account.balance < amount:
            raise ValueError("Недостатньо коштів на вихідному рахунку!")

        # Знімаємо гроші з одного рахунку
        from_account.balance -= amount
        # Додаємо гроші на інший рахунок
        to_account.balance += amount

        # Фіксуємо зміни
        session.commit()
        print("Переказ успішно виконано!")
    except Exception as e:
        # У випадку помилки відміняємо зміни
        session.rollback()
        print(f"Помилка виконання переказу: {e}")
    finally:
        # Закриваємо сесію
        session.close()

Ось так виглядає базова реалізація переказу коштів з використанням транзакції. Ми фіксуємо зміни через commit() тільки у випадку успішного виконання всіх кроків. Якщо щось пішло не так, використовуємо rollback().

Обробка помилок і відкат транзакцій

Помилки — це невід'ємна частина життя розробника. На щастя, транзакції дозволяють написати код так, щоб дані залишалися в консистентному стані навіть у разі збоїв. Ми можемо додати додаткову обробку помилок для більш детального контролю:

def transfer_money_safe(session: Session, from_account_id: int, to_account_id: int, amount: float):
    try:
        from_account = session.query(Account).filter(Account.id == from_account_id).with_for_update().first()
        to_account = session.query(Account).filter(Account.id == to_account_id).with_for_update().first()

        if from_account.balance < amount:
            raise ValueError("Недостатньо коштів!")

        from_account.balance -= amount
        to_account.balance += amount

        session.commit()
        print("Переказ виконано успішно!")
    except ValueError as ve:
        session.rollback()
        print(f"Помилка валідації: {ve}")
    except IntegrityError as ie:
        session.rollback()
        print(f"Помилка цілісності: {ie}")
    except Exception as e:
        session.rollback()
        print(f"Несподівана помилка: {e}")
    finally:
        session.close()

Тут використовується with_for_update() для блокування рядків до завершення транзакції. Це допоможе уникнути одночасної модифікації одних і тих самих даних різними процесами.

Практичні приклади

Тепер давайте перенесемо наш приклад у FastAPI і уявимо, що додаток має виконувати асинхронні операції.

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select

app = FastAPI()

@app.post("/transfer/")
async def transfer_money_api(
    from_account_id: int, to_account_id: int, amount: float, session: AsyncSession = Depends(get_db)
):
    try:
        async with session.begin():
            result_from = await session.execute(select(Account).where(Account.id == from_account_id))
            from_account = result_from.scalars().first()

            result_to = await session.execute(select(Account).where(Account.id == to_account_id))
            to_account = result_to.scalars().first()

            if not from_account or not to_account:
                raise HTTPException(status_code=404, detail="Акаунт не знайдено!")

            if from_account.balance < amount:
                raise HTTPException(status_code=400, detail="Недостатньо коштів!")

            from_account.balance -= amount
            to_account.balance += amount

            # Зміни фіксуються автоматично завдяки session.begin()
        return {"message": "Переказ виконано успішно!"}
    except Exception as e:
        await session.rollback()
        raise HTTPException(status_code=500, detail=f"Помилка переказу: {str(e)}")

Тут ми використовуємо async with session.begin() для роботи з транзакцією в асинхронному режимі. Це автоматизує відкриття і фіксацію транзакції, що значно спрощує код.

Важливі нюанси при роботі з транзакціями

1. Використання rollback: ніколи не забувайте відміняти транзакцію при виникненні помилки, інакше вона може зависнути.
2. Закриття сесії: після роботи з транзакцією обов'язково закривайте сесію, використовуючи session.close() (або автоматично за допомогою менеджера контексту).
3. Асинхронні сесії: якщо ви працюєте в асинхронному контексті, використовуйте async with session.begin() для коректної роботи.

Отже, транзакції та механізм відкатів (rollback) дозволяють захистити дані вашого додатку від помилок і збоїв у роботі. Ці знання знадобляться вам не тільки в банкінгу, але й у будь-яких проєктах, де потрібна атомарність операцій, таких як системи бронювання, e-commerce і інші високонавантажені сервіси.