Як працювати з транзакціями в FastAPI і Django через SQLAlchemy

Транзакції потрібні, щоб мінімізувати ризик неконсистентності даних. Наприклад, при виконанні складної бізнес-логіки, де зміна однієї таблиці залежить від змін в іншій.

Приклад з реального життя:

• Ви робите банківський переказ грошей від одного користувача іншому.
• Це включає два дії:
  1. Списання коштів з одного рахунку.
  2. Зарахування на інший рахунок.
• Якщо одна з операцій завершиться помилкою, транзакція має відкотити зміни, щоб дані лишилися консистентними.

---

Транзакції в FastAPI через SQLAlchemy

В екосистемі FastAPI ми можемо ефективно працювати з транзакціями, використовуючи SQLAlchemy і асинхронний підхід. Давай подивимось, як це зробити на практиці.

Крок 1: встановити базову конфігурацію проєкту (напевно в тебе вже давно все налаштовано, але буває всяке)

Якщо ти не налаштував SQLAlchemy для роботи з FastAPI, зроби це спочатку. Нагадаю коротко:

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from fastapi import FastAPI, Depends
from sqlalchemy.ext.declarative import declarative_base

DATABASE_URL = "postgresql+asyncpg://user:password@localhost/db"

engine = create_async_engine(DATABASE_URL, echo=True)
SessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()

# Dependency для отримання сесії
async def get_db():
    async with SessionLocal() as session:
        yield session

Крок 2: використовувати транзакції для виконання операцій

Уяви, що ми розробляємо невеликий API для керування банківськими переказами.

Приклад операції без транзакції:

from sqlalchemy.exc import SQLAlchemyError

async def transfer_funds(sender_id: int, recipient_id: int, amount: float, db: AsyncSession):
    try:
        sender = await db.execute(select(User).where(User.id == sender_id))
        recipient = await db.execute(select(User).where(User.id == recipient_id))

        sender.balance -= amount
        recipient.balance += amount

        # Збереження змін
        await db.commit()
    except SQLAlchemyError:
        # У разі помилки жодних змін у базі не буде
        await db.rollback()

Активне застосування транзакцій

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

Приклад:

async def transfer_funds_with_transaction(sender_id: int, recipient_id: int, amount: float, db: AsyncSession):
    try:
        async with db.begin():  # Початок транзакції
            sender = await db.execute(select(User).where(User.id == sender_id))
            recipient = await db.execute(select(User).where(User.id == recipient_id))

            sender.balance -= amount
            recipient.balance += amount

        # COMMIT буде викликано автоматично після завершення блоку
    except SQLAlchemyError:
        # Помилка призведе до автоматичного виклику ROLLBACK
        await db.rollback()
        raise

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

---

Транзакції в Django через SQLAlchemy

Інтеграція SQLAlchemy в Django — задача не дуже стандартна, але можлива. Проте в більшості випадків транзакції в Django обробляються його рідним ORM. Якщо ж ти все-таки використовуєш SQLAlchemy в Django, усе працює аналогічно.

from sqlalchemy.ext.asyncio import AsyncSession
from django.http import JsonResponse

# Django використовує middleware, можна обробляти транзакції вручну
async def transfer_funds_django_view(request):
    sender_id = request.POST.get("sender_id")
    recipient_id = request.POST.get("recipient_id")
    amount = float(request.POST.get("amount"))

    async with AsyncSession() as session:
        try:
            async with session.begin():  # Початок транзакції
                sender = await session.execute(select(User).where(User.id == sender_id))
                recipient = await session.execute(select(User).where(User.id == recipient_id))

                sender.balance -= amount
                recipient.balance += amount

            # Транзакція успішно завершена
            return JsonResponse({"status": "success"})
        except SQLAlchemyError:
            await session.rollback()
            return JsonResponse({"status": "error", "message": "Transaction failed"})

---

Робота з асинхронними операціями

Асинхронність додає своїх нюансів, особливо коли мова про транзакції. Наприклад, при виконанні довгих операцій (наприклад, зовнішніх HTTP-запитів) всередині транзакції може виникнути тайм-аут.

Поради щоб уникнути проблем з асинхронними транзакціями:

1. Мінімізуйте час виконання транзакції: операції, не пов'язані з базою даних (наприклад, виклики зовнішніх API), виконуй до або після транзакції.
2. Використовуйте timeouts: для довгих операцій виставляйте розумний тайм-аут на рівні движка бази даних.
3. Тестуйте граничні сценарії: перевір, як транзакції поводяться при неочевидних помилках (наприклад, недоступність бази даних).

---

Практичний приклад: транзакції в FastAPI

Припустимо, ми будуємо API для обробки замовлень в магазині. Нам треба:

1. Спочатку перевірити, чи є товар на складі.
2. Потім зменшити кількість товару.
3. Після цього створити запис про нове замовлення.

Код:

from sqlalchemy.exc import SQLAlchemyError

@router.post("/create_order/")
async def create_order(order: OrderCreate, db: AsyncSession = Depends(get_db)):
    try:
        async with db.begin():
            product = await db.execute(select(Product).where(Product.id == order.product_id))
            product = product.scalar_one_or_none()

            if not product or product.stock < order.quantity:
                raise HTTPException(status_code=400, detail="Product out of stock")

            product.stock -= order.quantity
            new_order = Order(customer_id=order.customer_id, product_id=order.product_id, quantity=order.quantity)
            db.add(new_order)

        # COMMIT станеться автоматично, якщо все пройде успішно
        return {"status": "success"}
    except SQLAlchemyError as e:
        await db.rollback()
        return {"status": "error", "message": str(e)}

---

Особливості роботи з транзакціями в Django ORM (наприклад, atomic)

Якщо ти використовуєш вбудований Django ORM, транзакції обробляються через декоратор transaction.atomic. Ось як це виглядає:

from django.db import transaction

@transaction.atomic
def transfer_funds(sender_id, recipient_id, amount):
    sender = User.objects.get(id=sender_id)
    recipient = User.objects.get(id=recipient_id)

    sender.balance -= amount
    recipient.balance += amount

    sender.save()
    recipient.save()

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

---

Типові помилки при роботі з транзакціями

1. Забування про rollback: іноді розробники забувають викликати rollback в місцях, де транзакція падає з помилкою. Це може призвести до "висілих" транзакцій в базі.
2. Ігнорування виключень: ніколи не ігноруй помилки — транзакція може залишитись "підвішеною".
3. Довгі операції в транзакції: не роби довгих операцій (наприклад, зовнішніх HTTP-запитів) всередині блоку транзакції.

---

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