SQLAlchemy 2.0 — ORM, Core та Async Engine

Жоден production-рівень API не живе у вакуумі. Будь-який реальний вебсервіс неминуче стикається з необхідністю зберігати, читати, оновлювати та видаляти дані у постійному сховищі — реляційній базі даних. Саме тут на сцену виходить SQLAlchemy — найповніший, найзріліший і найпопулярніший інструментарій для роботи з базами даних у Python, який часто називають «швейцарським ножем» між ORM та SQL-будівником.

Проте SQLAlchemy — це не просто ще один ORM. Це ціла екосистема, що складається з двох чітко розмежованих рівнів: Core (низькорівневий будівник SQL-запитів) і ORM (об'єктно-реляційне відображення). Знання обох рівнів і розуміння того, де закінчується один і починається інший, є ознакою зрілого Python-розробника.

У версії 2.0, яка вийшла у лютому 2023 року після тривалого перехідного періоду, SQLAlchemy отримала принципово новий, сучасний Python-синтаксис із підтримкою повної статичної типізації. Якщо ви бачили старий стиль SQLAlchemy 1.x — із класичними декларативними моделями на базі Column() та рядковими рефернесами — забудьте про нього. SQLAlchemy 2.0 виглядає і відчувається зовсім по-іншому. Власне, саме для цього і написана ця стаття.

«Початок з далека»: Навіщо взагалі потрібен ORM?

Уявіть, що ви пишете FastAPI-ендпоінт, який має отримати список проектів конкретного користувача з бази даних PostgreSQL. Без жодного ORM або допоміжного інструменту вам доведеться:

1. Відкрити низькорівневе підключення до бази даних через Python-драйвер (psycopg2 або asyncpg).
2. Сформувати SQL-рядок вручну — потенційно вразливий до SQL-ін'єкцій.
3. Виконати запит і отримати сирий результат у вигляді списку кортежів або словників.
4. Вручну перетворити ці дані у Python-об'єкти (або Pydantic-моделі).
5. При оновленні — стежити за тим, які поля змінилися, і самостійно генерувати UPDATE-запит.
6. Явно керувати транзакціями: BEGIN, COMMIT, ROLLBACK.

Це величезний обсяг шаблонного коду (boilerplate), схильного до помилок і складного у тестуванні. ORM (Object-Relational Mapper) вирішує цю проблему, надаючи абстракцію між реляційним світом таблиць і рядків та об'єктно-орієнтованим світом Python-класів і екземплярів.

@startuml
skinparam style plain
skinparam linetype ortho
skinparam defaultFontSize 13
skinparam backgroundColor #ffffff

package "SQLAlchemy Ecosystem" {
    package "ORM Layer (Об'єктний рівень)" #e8f4f8 {
        component [DeclarativeBase\nMapped Classes] as models
        component [Session\nAsyncSession] as session
        component [Unit of Work\nIdentity Map] as uow
        component [Relationships\nLazy/Eager Loading] as rel
    }

    package "Core Layer (SQL рівень)" #fff3cd {
        component [Engine\nAsyncEngine] as engine
        component [Connection Pool] as pool
        component [select() insert()\nupdate() delete()] as dml
        component [Table\nColumn\nMetaData] as meta
    }

    package "DBAPI (Драйвер)" #f8d7da {
        component [psycopg2 / psycopg\n(Sync PostgreSQL)] as sync_drv
        component [asyncpg\n(Async PostgreSQL)] as async_drv
    }

    database "PostgreSQL" as pg
}

models --> session
session --> uow
uow --> dml
session --> dml
dml --> engine
engine --> pool
pool --> sync_drv
pool --> async_drv
sync_drv --> pg
async_drv --> pg
meta --> dml

@enduml

Розглянемо кожен рівень детально.

SQLAlchemy Core — «Конструктор SQL»

Core — це нижній рівень SQLAlchemy. Він надає Python-API для побудови SQL-виразів як об'єктів мови Python, а не як рядків. Цей рівень є «мовонезалежним» у розумінні бази даних: ви будуєте вираз select(users_table).where(users_table.c.id == 42), а SQLAlchemy сама генерує правильний SQL-діалект для PostgreSQL, MySQL, SQLite чи Oracle.

Ключові компоненти Core:

• Engine — «серце» SQLAlchemy, яке управляє пулом підключень до бази даних.
• MetaData та Table — опис структури таблиць як Python-об'єктів.
• DML-функції — select(), insert(), update(), delete() для побудови запитів.
• Connection — об'єкт одного активного підключення до БД.

SQLAlchemy ORM — «Об'єктне відображення»

ORM — це верхній рівень, побудований поверх Core. Він надає класичне об'єктно-реляційне відображення: ви працюєте з Python-класами (моделями), а SQLAlchemy автоматично перетворює операції над ними у відповідні SQL-запити.

Ключові компоненти ORM:

• DeclarativeBase та Mapped — оголошення моделей (таблиць) як Python-класів.
• Session та AsyncSession — «workspace» для роботи з об'єктами, реалізація патерну Unit of Work.
• relationship() — декларативний опис зв'язків між таблицями (one-to-many, many-to-many тощо).
• Identity Map — внутрішній реєстр, що гарантує унікальність об'єктів у межах однієї сесії.

Порівняння: EF Core ↔ SQLAlchemy

Перед тим як занурюватися в код, варто провести паралелі з інструментарієм ASP.NET, щоб сформувати правильні ментальні моделі. Якщо ви вже працювали з Entity Framework Core, більшість концепцій SQLAlchemy одразу стануть інтуїтивно зрозумілими — назви відрізняються, але ідеї залишаються тими самими.

Найважливіша відмінність: в EF Core DbContext є одночасно і «сховищем об'єктів», і точкою доступу до запитів (через DbSet). У SQLAlchemy ці ролі чітко розділені: Engine управляє підключеннями, а Session управляє об'єктами. Це розділення є більш явним і дозволяє краще контролювати життєвий цикл обох сутностей.

Engine та Connection Pool: Фундамент SQLAlchemy

Engine — це перший об'єкт, який потрібно створити для роботи з SQLAlchemy. Він є центральною точкою конфігурації та управляє пулом підключень (connection pool) до бази даних.

Чому потрібен Connection Pool?

Відкриття нового підключення до PostgreSQL — це відносно «дорога» операція: вона включає TCP-рукостискання, автентифікацію та ініціалізацію внутрішнього стану сесії PostgreSQL. У вебзастосунку, де одночасно можуть надходити сотні або тисячі запитів, відкривати і закривати з'єднання для кожного запиту є неприпустимо повільним підходом.

Connection Pool вирішує цю проблему: він заздалегідь відкриває кілька підключень і тримає їх «живими». Коли FastAPI-обробнику потрібне підключення до бази даних, він бере готове з пулу, використовує його, і повертає назад — підключення не закривається, а очікує наступного запиту.

@startuml
skinparam style plain
skinparam linetype ortho
skinparam defaultFontSize 13
skinparam ArrowColor #555555
skinparam backgroundColor #ffffff

package "FastAPI Application" #e8f4f8 {
    component [Request 1] as R1
    component [Request 2] as R2
    component [Request 3] as R3
}

package "Connection Pool (QueuePool)" #fff3cd {
    component [Connection 1\n(зайнята)] as C1 #ffcccc
    component [Connection 2\n(вільна)]  as C2 #ccffcc
    component [Connection 3\n(вільна)]  as C3 #ccffcc
    component [Connection 4\n(overflow)] as C4 #ffe0b2
}

database "PostgreSQL Server" as PG #f8d7da

R1 --> C1
R2 --> C2
R3 --> C3

C1 --> PG
C2 --> PG
C3 --> PG
C4 .right.> PG : створюється при\nперевантаженні

@enduml

Python DB-API 2.0: Що таке «драйвер» і як він працює?

Перш ніж розбирати create_engine(), важливо зрозуміти рівень, що знаходиться нижче SQLAlchemy, — безпосередній Python-драйвер для PostgreSQL.

DB-API 2.0 (PEP 249) — це стандартний Python-інтерфейс для роботи з реляційними базами даних. Він визначає мінімальний набір функцій і об'єктів, які має реалізувати будь-який Python-пакет для роботи з конкретною СУБД. Завдяки цьому стандарту SQLAlchemy може прозоро перемикатися між різними драйверами без зміни вашого коду.

Аналогія: DB-API 2.0 — це «стандартна розетка», а конкретні драйвери (psycopg2, asyncpg, psycopg) — це «прилади», що підключаються до неї. SQLAlchemy — це «подовжувач», який надає вам зручний API поверх будь-якого «приладу».

Як виглядає робота з драйвером напряму?

Щоб по-справжньому оцінити, від чого абстрагує SQLAlchemy, подивимося на «голий» DB-API 2.0 з psycopg2:

import psycopg2
from psycopg2.extras import RealDictCursor

# 1. Відкриваємо підключення вручну
conn = psycopg2.connect(
    host="localhost",
    port=5432,
    dbname="taskforge_db",
    user="user",
    password="password",
)

try:
    # 2. Створюємо курсор (об'єкт для виконання запитів)
    # RealDictCursor → результати у вигляді dict замість tuple
    with conn.cursor(cursor_factory=RealDictCursor) as cur:

        # 3. Виконуємо параметризований запит (захист від SQL-ін'єкцій)
        cur.execute(
            "SELECT id, username, email FROM users WHERE is_active = %s LIMIT %s",
            (True, 10),  # Параметри — окремим кортежем, НЕ через f-string!
        )

        # 4. Отримуємо результати
        rows = cur.fetchall()
        for row in rows:
            print(row["id"], row["username"])  # row — це dict

        # 5. INSERT з поверненням згенерованого id
        cur.execute(
            "INSERT INTO users (username, email) VALUES (%s, %s) RETURNING id",
            ("new_user", "new@example.com"),
        )
        new_id = cur.fetchone()["id"]

    # 6. Підтверджуємо транзакцію вручну
    conn.commit()

except Exception:
    conn.rollback()  # Відкат при будь-якій помилці
    raise
finally:
    conn.close()  # Завжди закриваємо підключення

Цей код працює, але очевидно, наскільки він багатослівний: ручне управління з'єднаннями, курсорами, транзакціями та перетворенням результатів — все це ваша відповідальність. SQLAlchemy автоматизує весь цей boilerplate.

Асинхронний аналог із asyncpg виглядає інакше (asyncpg не реалізує DB-API 2.0, а має власний API, оптимізований для asyncio):

import asyncpg

async def main():
    # asyncpg підключається напряму без курсорів
    conn = await asyncpg.connect(
        host="localhost", port=5432,
        database="taskforge_db", user="user", password="password",
    )
    try:
        # asyncpg повертає список Record-об'єктів (схожі на dict)
        rows = await conn.fetch(
            "SELECT id, username FROM users WHERE is_active = $1 LIMIT $2",
            True, 10,  # Параметри через $1, $2 (PostgreSQL-стиль)
        )
        for row in rows:
            print(row["id"], row["username"])

        new_id = await conn.fetchval(
            "INSERT INTO users (username, email) VALUES ($1, $2) RETURNING id",
            "new_user", "new@example.com",
        )
    finally:
        await conn.close()

Які драйвери обирати у 2026+ році?

На сьогодні для PostgreSQL існує три основних варіанти драйверів. Ось актуальна картина:

pip install asyncpg

pip install "psycopg[binary]"

pip install sqlalchemy asyncpg

Які бази даних підтримує SQLAlchemy «з коробки»?

SQLAlchemy не прив'язана до PostgreSQL. Офіційна документація визначає п'ять вбудованих (included) діалектів — тобто БД, для яких підтримка вбудована безпосередньо в пакет sqlalchemy без встановлення сторонніх розширень.

Розглянемо кожну БД детальніше.

MySQL та MariaDB

MySQL та його повністю сумісний форк MariaDB є найпоширенішими реляційними СУБД у веброзробці. SQLAlchemy підтримує обидві через однаковий mysql+<driver>:// префікс (MariaDB автоматично визначається за версією сервера).

# mysqlclient: найшвидший синхронний драйвер для MySQL (C-розширення)
# pip install mysqlclient
engine = create_engine(
    "mysql+mysqldb://user:password@localhost:3306/mydb?charset=utf8mb4"
)

# asyncmy: сучасний async-драйвер для MySQL/MariaDB
# pip install asyncmy
async_engine = create_async_engine(
    "mysql+asyncmy://user:password@localhost:3306/mydb?charset=utf8mb4"
)

# PyMySQL: pure-Python драйвер, простий у встановленні (без компіляції)
# pip install pymysql
engine = create_engine(
    "mysql+pymysql://user:password@localhost:3306/mydb?charset=utf8mb4"
)

SQLite

SQLite — вбудована у Python (import sqlite3) легковажна СУБД без сервера: вся база даних зберігається в одному .db-файлі на диску. Ідеально підходить для локальної розробки, тестів та невеликих проєктів.

# Не потрібен додатковий пакет — sqlite3 вбудований у Python
engine = create_engine(
    "sqlite:///./taskforge_dev.db"  # відносний шлях до файлу
)

# In-memory БД (існує лише у RAM, зникає при закритті):
engine_memory = create_engine("sqlite:///:memory:")

# aiosqlite: async-обгортка над sqlite3
# pip install aiosqlite
async_engine = create_async_engine(
    "sqlite+aiosqlite:///./taskforge_dev.db"
)

# Async in-memory:
async_engine_memory = create_async_engine("sqlite+aiosqlite:///:memory:")

Microsoft SQL Server (MSSQL)

Microsoft SQL Server широко використовується у корпоративному середовищі (особливо там, де вже є інфраструктура Microsoft). SQLAlchemy підтримує його через ODBC-підключення.

# pip install pyodbc
# Вимагає встановленого ODBC Driver for SQL Server на ОС
engine = create_engine(
    "mssql+pyodbc://user:password@mssql-server:1433/mydb"
    "?driver=ODBC+Driver+18+for+SQL+Server"
    "&TrustServerCertificate=yes"
)

# pip install aioodbc
from sqlalchemy.ext.asyncio import create_async_engine

async_engine = create_async_engine(
    "mssql+aioodbc://user:password@mssql-server:1433/mydb"
    "?driver=ODBC+Driver+18+for+SQL+Server"
    "&TrustServerCertificate=yes"
)

Зведена таблиця вибору БД та драйвера для нових проєктів

Синхронний Engine: create_engine()

Функція create_engine() приймає рядок підключення (DSN — Data Source Name) та низку опцій, що конфігурують поведінку пулу.

from sqlalchemy import create_engine

# Рядок підключення (DSN) для PostgreSQL з psycopg2
DATABASE_URL = "postgresql+psycopg2://user:password@localhost:5432/taskforge_db"

engine = create_engine(
    DATABASE_URL,
    # --- Параметри Connection Pool ---
    pool_size=10,       # Кількість «постійних» підключень у пулі
    max_overflow=20,    # Максимум тимчасових підключень понад pool_size
    pool_timeout=30,    # Секунд очікування вільного підключення з пулу
    pool_recycle=1800,  # Перестворювати підключення кожні 30 хвилин
                        # (щоб уникнути "stale connections" від PostgreSQL)
    # --- Налагодження ---
    echo=True,          # Виводити всі SQL-запити у консоль (лише для розробки!)
)

Асинхронний Engine: create_async_engine()

Для роботи з FastAPI у повністю асинхронному режимі (що є рекомендованим підходом, як ми дізналися у статті 14) необхідно використовувати create_async_engine() разом із асинхронним драйвером asyncpg.

from sqlalchemy.ext.asyncio import create_async_engine

# Зверніть на префікс: postgresql+asyncpg (не psycopg2!)
ASYNC_DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/taskforge_db"

async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_recycle=1800,
    echo=False,  # У production завжди False
)

Ключова різниця між create_engine() та create_async_engine() полягає у внутрішньому DBAPI-драйвері:

• postgresql+psycopg2 → синхронний драйвер, блокує потік виконання.
• postgresql+asyncpg → асинхронний драйвер, звільняє event loop Python під час очікування відповіді від БД.

Session та sessionmaker: Одиниця роботи

Session — це центральний об'єкт для роботи з ORM. Він реалізує патерн Unit of Work (Одиниця роботи), що є фундаментальним архітектурним патерном для роботи з базами даних (ми детально розглянемо його у розділі «Під капотом»).

Коротко: Session — це «записник» або «тимчасова пам'ять», яка:

1. Відстежує всі Python-об'єкти (ORM-моделі), завантажені з БД або додані до нього.
2. Накопичує всі зміни (INSERT, UPDATE, DELETE), не надсилаючи їх одразу до БД.
3. При виклику session.flush() або session.commit() — генерує відповідні SQL-запити та виконує їх у межах однієї транзакції.

sessionmaker та async_sessionmaker

Оскільки Session має бути короткоживучим об'єктом (один запит = одна сесія), зручно мати фабрику, яка створює нові сесії з потрібними налаштуваннями. Для цього слугують sessionmaker та async_sessionmaker.

Базовий шаблон виглядає так:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("postgresql+psycopg://user:pass@localhost/db")

SessionLocal = sessionmaker(
    bind=engine,
    autocommit=False,
    autoflush=False,
    expire_on_commit=False,
)

# Один HTTP-запит = одна сесія:
with SessionLocal() as session:
    user = session.get(User, 1)
    user.is_active = False
    session.commit()

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

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

AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    autocommit=False,
    autoflush=False,
    expire_on_commit=False,
)

# Один HTTP-запит = одна async-сесія:
async with AsyncSessionLocal() as session:
    user = await session.get(User, 1)
    user.is_active = False
    await session.commit()

Синхронний Engine: create_engine()

Функція create_engine() приймає рядок підключення (DSN — Data Source Name) та низку опцій, що конфігурують поведінку пулу.

from sqlalchemy import create_engine

# Рядок підключення (DSN) для PostgreSQL з psycopg2
DATABASE_URL = "postgresql+psycopg2://user:password@localhost:5432/taskforge_db"

engine = create_engine(
    DATABASE_URL,
    # --- Параметри Connection Pool ---
    pool_size=10,       # Кількість «постійних» підключень у пулі
    max_overflow=20,    # Максимум тимчасових підключень понад pool_size
    pool_timeout=30,    # Секунд очікування вільного підключення з пулу
    pool_recycle=1800,  # Перестворювати підключення кожні 30 хвилин
                        # (щоб уникнути "stale connections" від PostgreSQL)
    # --- Налагодження ---
    echo=True,          # Виводити всі SQL-запити у консоль (лише для розробки!)
)

Асинхронний Engine: create_async_engine()

Для роботи з FastAPI у повністю асинхронному режимі (що є рекомендованим підходом, як ми дізналися у статті 14) необхідно використовувати create_async_engine() разом із асинхронним драйвером asyncpg.

from sqlalchemy.ext.asyncio import create_async_engine

# Зверніть на префікс: postgresql+asyncpg (не psycopg2!)
ASYNC_DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/taskforge_db"

async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_recycle=1800,
    echo=False,  # У production завжди False
)

Ключова різниця між create_engine() та create_async_engine() полягає у внутрішньому DBAPI-драйвері:

• postgresql+psycopg2 → синхронний драйвер, блокує потік виконання.
• postgresql+asyncpg → асинхронний драйвер, звільняє event loop Python під час очікування відповіді від БД.

Параметри конфігурації Connection Pool у Engine

При створенні Engine або AsyncEngine за допомогою параметрів можна тонко налаштувати поведінку вбудованого пулу з'єднань (QueuePool). Розглянемо ключові параметри, їхні значення за замовчуванням та вплив на систему:

• pool_size (тип int, за замовчуванням 5) Визначає кількість «постійно відкритих» (keep-alive) підключень, які пул зберігає в пам'яті. Ці з'єднання ніколи не закриваються при звільненні.
  • На що впливає: Лімітує базову пропускну здатність. Якщо ваші FastAPI-воркери роблять мало паралельних запитів, 5–10 підключень цілком достатньо. Для навантажених додатків це значення збільшують до 20–50.
  • Застереження: Кожне з'єднання споживає RAM на сервері PostgreSQL (близько 10 МБ на з'єднання). Збільшення pool_size понад ліміт max_connections у конфігурації PostgreSQL призведе до того, що база даних відхилятиме нові підключення.
• max_overflow (тип int, за замовчуванням 10) Визначає максимальну кількість тимчасових додаткових підключень, які пул може створити, якщо всі з'єднання з pool_size зараз зайняті.
  • На що впливає: Дозволяє додатку безболісно переживати раптові піки навантаження. Коли пік минає, ці «overflow» підключення автоматично закриваються.
  • Розрахунок: Загальна максимальна кількість одночасно відкритих підключень з одного екземпляра додатка дорівнює pool_size + max_overflow. Наприклад, pool_size=10, max_overflow=20 дає ліміт у 30 з'єднань.
• pool_recycle (тип int, за замовчуванням -1, тобто вимкнено) Визначає максимальний вік підключення у секундах. Якщо підключення старше за це значення, при поверненні в пул воно буде закрите, а замість нього буде створено нове.
  • На що впливає: Рятує від проблеми stale connections (застарілих підключень). Деякі мережеві екрани (firewalls), хмарні бази даних (наприклад, AWS RDS, Heroku Postgres) або сам PostgreSQL можуть примусово розривати неактивні TCP-сесії після певного періоду простою (наприклад, 10 або 30 хвилин). Якщо пул спробує використати таке з'єднання, виникне помилка на кшталт OperationalError: Connection handshake failed. Встановлення pool_recycle=1800 (30 хвилин) змушує оновлювати підключення превентивно.
• pool_pre_ping (тип bool, за замовчуванням False) Якщо встановлено в True, при кожному отриманні з'єднання з пулу SQLAlchemy виконує швидкий тестовий запит (зазвичай SELECT 1) перед тим, як віддати з'єднання вашому коду.
  • На що впливає: Забезпечує високу стійкість до збоїв. Якщо з'єднання виявиться «мертвим» (через перезапуск бази даних або мережевий збій), пул тихо закриє його, відкриє нове працездатне підключення та прозоро віддасть вашому коду. Розробник навіть не помітить, що з'єднання розривалося.
  • Рекомендація: Завжди ставте pool_pre_ping=True у production. Хоча це додає крихітний оверхед на один SQL-запит-тест, стійкість додатку зростає в рази.

Приклад повної production конфігурації Engine:

async_engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,           # Збільшено для навантаженого API
    max_overflow=10,        # Додаткові 10 з'єднань під час піків
    pool_recycle=1800,      # Оновлювати кожні 30 хвилин (захист від AWS RDS таймаутів)
    pool_pre_ping=True,     # ✅ Перевіряти працездатність з'єднання перед відправкою запиту
)

Session та sessionmaker: Одиниця роботи

Session — це центральний об'єкт для роботи з ORM. Він реалізує патерн Unit of Work (Одиниця роботи), що є фундаментальним архітектурним патерном для роботи з базами даних (ми детально розглянемо його у розділі «Під капотом»).

Коротко: Session — це «записник» або «тимчасова пам'ять», яка:

1. Відстежує всі Python-об'єкти (ORM-моделі), завантажені з БД або додані до нього.
2. Накопичує всі зміни (INSERT, UPDATE, DELETE), не надсилаючи їх одразу до БД.
3. При виклику session.flush() або session.commit() — генерує відповідні SQL-запити та виконує їх у межах однієї транзакції.

sessionmaker та async_sessionmaker

Оскільки Session має бути короткоживучим об'єктом (один запит = одна сесія), зручно мати фабрику, яка створює нові сесії з потрібними налаштуваннями. Для цього слугують sessionmaker та async_sessionmaker.

Базовий шаблон виглядає так:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("postgresql+psycopg://user:pass@localhost/db")

SessionLocal = sessionmaker(
    bind=engine,
    autocommit=False,
    autoflush=False,
    expire_on_commit=False,
)

# Один HTTP-запит = одна сесія:
with SessionLocal() as session:
    user = session.get(User, 1)
    user.is_active = False
    session.commit()

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

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

AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    autocommit=False,
    autoflush=False,
    expire_on_commit=False,
)

# Один HTTP-запит = одна async-сесія:
async with AsyncSessionLocal() as session:
    user = await session.get(User, 1)
    user.is_active = False
    await session.commit()

Параметри sessionmaker та async_sessionmaker: повний розбір

Обидві фабрики приймають однакові параметри. Нижче — вичерпна документація кожного з них із прикладами впливу на поведінку.

bind / autobegin

Прив'язує фабрику до конкретного Engine або AsyncEngine. Усі сесії, створені цією фабрикою, будуть автоматично використовувати саме це підключення до БД. Якщо не вказати тут — доведеться передавати bind при кожному виклику Session().

Якщо True, SQLAlchemy автоматично починає нову транзакцію (BEGIN) при першій операції з БД. Якщо False — транзакцію потрібно починати вручну через session.begin(). Значення True (за замовчуванням) є зручним та безпечним для більшості випадків.

# Стандартне використання — bind + autobegin=True (за замовчуванням)
SessionLocal = sessionmaker(bind=engine, autobegin=True)

with SessionLocal() as session:
    # При першому execute() SQLAlchemy автоматично відкриває транзакцію:
    # BEGIN (implicit)
    user = session.get(User, 1)
    session.commit()
    # COMMIT

# autobegin=False — повний ручний контроль (рідко потрібен):
ManualSession = sessionmaker(bind=engine, autobegin=False)
with ManualSession() as session:
    with session.begin():         # Явний BEGIN
        user = session.get(User, 1)
    # COMMIT автоматично при виході з session.begin()

autocommit

Чому False? Це один з найважливіших параметрів, що забезпечує атомарність операцій.

Якщо autocommit=True — кожен SQL-запит виконується у своїй власній транзакції і негайно комітиться. Це означає, що неможливо атомарно виконати кілька операцій: якщо після першого INSERT програма впаде — другий INSERT вже буде збережений у БД, а перший — ні. Дані опиняться у неузгодженому стані.

Якщо autocommit=False — всі операції в межах сесії виконуються в одній транзакції, яку ви контролюєте через commit() / rollback(). Це гарантує ACID-властивості.

# ❌ autocommit=True — небезпечно, не використовуйте у production
AutoCommitSession = sessionmaker(bind=engine, autocommit=True)

with AutoCommitSession() as session:
    # Кожен виклик одразу комітиться окремо!
    session.add(User(username="alice"))  # SQL: INSERT + COMMIT
    # Якщо тут виникне помилка:
    raise RuntimeError("Щось пішло не так!")
    session.add(User(username="bob"))    # Цей INSERT ніколи не виконається
    # alice вже збережена у БД — неузгодженість!

# ✅ autocommit=False — безпечно (за замовчуванням)
SafeSession = sessionmaker(bind=engine, autocommit=False)

with SafeSession() as session:
    try:
        session.add(User(username="alice"))
        session.add(User(username="bob"))
        session.commit()  # Обидва INSERT у ONE транзакції
    except Exception:
        session.rollback()  # Обидва відкочуються — БД у чистому стані

autoflush

Чому у FastAPI зазвичай ставлять False?

За замовчуванням autoflush=True: SQLAlchemy автоматично виконує flush() (відправляє накопичені зміни у БД у рамках поточної транзакції) перед кожним SELECT-запитом у межах поточної сесії.

На перший погляд це зручно: ви завжди бачите актуальні дані навіть до commit(). Але у FastAPI-сервісах, де сесія часто передається через DI між кількома шарами (router → service → repository), autoflush=True може призводити до неочікуваних SQL-запитів і ускладнює налагодження.

Рекомендація для FastAPI: autoflush=False. Ви самостійно викликаєте session.flush() тільки коли це потрібно (наприклад, щоб отримати згенерований id). Це дає повний контроль і передбачувану поведінку.

# ✅ autoflush=True (за замовчуванням у SQLAlchemy)
AutoFlushSession = sessionmaker(bind=engine, autoflush=True)

with AutoFlushSession() as session:
    new_user = User(username="charlie")
    session.add(new_user)
    # ↓ Тут ще немає flush()

    # Але при будь-якому SELECT — автоматичний flush перед ним:
    existing = session.execute(select(User).where(User.username == "alice")).scalar()
    # SQLAlchemy робить:
    # 1. INSERT INTO users (username) VALUES ('charlie')  ← autoflush!
    # 2. SELECT * FROM users WHERE username = 'alice'

    # Це зручно, але може бути несподіваним, якщо charlie ще «не готовий»
    # (наприклад, не встановлено обов'язкове поле email)

# ✅ autoflush=False (рекомендовано для FastAPI)
ManualFlushSession = sessionmaker(bind=engine, autoflush=False)

with ManualFlushSession() as session:
    new_user = User(username="charlie")
    session.add(new_user)

    # SELECT виконується одразу, без autoflush
    existing = session.execute(select(User).where(User.username == "alice")).scalar()
    # SQL: SELECT * FROM users WHERE username = 'alice'
    # charlie ще не INSERT-нутий — SELECT не «бачить» його

    # Явний flush лише тоді, коли потрібен id:
    session.flush()  # INSERT INTO users ...
    print(new_user.id)  # Тепер id доступний

    session.commit()

expire_on_commit

Найважливіший параметр для async FastAPI — завжди ставте False!

Після виклику commit() SQLAlchemy за замовчуванням (True) «відмічає» всі об'єкти у сесії як expired (застарілі). Це означає: при наступному зверненні до будь-якого атрибуту об'єкта SQLAlchemy виконає новий SELECT-запит, щоб «освіжити» дані з БД.

У синхронному режимі це відбувається непомітно — lazy refresh є автоматичним.

В асинхронному режимі це катастрофа: після await session.commit() сесія більше не активна в async-контексті. Спроба отримати атрибут expired-об'єкта викличе MissingGreenlet: greenlet_spawn has not been called — одна з найпоширеніших помилок початківців у async SQLAlchemy.

# --- SYNC: expire_on_commit=True (за замовчуванням) ---
SyncSession = sessionmaker(bind=engine, expire_on_commit=True)

with SyncSession() as session:
    user = session.get(User, 1)
    user.username = "updated"
    session.commit()
    # user тепер expired!

    # Але sync-режим «рятує» нас автоматично:
    print(user.username)
    # SELECT * FROM users WHERE id = 1  ← автоматичний lazy SELECT!
    # Виводить: "updated"   ✅ Працює, але виконує зайвий запит

# --- ASYNC: expire_on_commit=True — ПОМИЛКА! ---
BadAsyncSession = async_sessionmaker(bind=async_engine, expire_on_commit=True)

async with BadAsyncSession() as session:
    user = await session.get(User, 1)
    user.username = "updated"
    await session.commit()
    # user тепер expired!

    print(user.username)
    # ❌ MissingGreenlet: greenlet_spawn has not been called
    # Немає активної async-транзакції для lazy SELECT!

# --- ASYNC: expire_on_commit=False — правильно! ---
GoodAsyncSession = async_sessionmaker(bind=async_engine, expire_on_commit=False)

async with GoodAsyncSession() as session:
    user = await session.get(User, 1)
    user.username = "updated"
    await session.commit()
    # user НЕ expired — значення в пам'яті збережено

    print(user.username)  # ✅ "updated" — з пам'яті Python, без SQL

class_ (лише для async_sessionmaker)

Визначає конкретний клас, який буде створювати фабрика. Для async_sessionmaker потрібно явно передати class*=AsyncSession. Це дозволяє підставляти власні підкласи AsyncSession з кастомною логікою (наприклад, автоматичним логуванням або метриками).

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

class LoggedSession(AsyncSession):
    """AsyncSession з логуванням кожного коміту."""
    async def commit(self):
        print(f"[DB] Committing transaction in session {id(self)}")
        await super().commit()

LoggedSessionLocal = async_sessionmaker(
    bind=async_engine,
    class_=LoggedSession,  # Використовуємо наш кастомний клас
    expire_on_commit=False,
)

async with LoggedSessionLocal() as session:
    # ... операції ...
    await session.commit()
    # Виведе: [DB] Committing transaction in session 140234567890

Підсумковий конфіг для production FastAPI

Ось канонічна конфігурація async_sessionmaker для production FastAPI-додатку на основі всього вищесказаного:

from sqlalchemy.ext.asyncio import (
    create_async_engine,
    async_sessionmaker,
    AsyncSession,
)

async_engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost:5432/taskforge_db",
    pool_size=10,
    max_overflow=20,
    pool_recycle=1800,
    echo=False,
)

AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    autocommit=False,    # ✅ Явний контроль транзакцій (ACID-безпека)
    autoflush=False,     # ✅ Явний flush() — передбачувана поведінка
    expire_on_commit=False,  # ✅ Обов'язково для async FastAPI
)

ORM-Моделі у SQLAlchemy 2.0: Новий «Python-first» синтаксис

Якщо у вас є досвід роботи зі SQLAlchemy 1.x, перший погляд на код SQLAlchemy 2.0 може викликати легкий шок — він виглядає зовсім інакше. Це навмисна зміна: команда SQLAlchemy чітко взяла курс на максимальну інтеграцію з сучасним Python-синтаксисом типізації, наслідуючи ідеї Pydantic та dataclasses.

Еволюція синтаксису: від 1.x до 2.0

Щоб оцінити масштаб змін, подивимося на одну і ту саму модель у різних синтаксисах:

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

Base = declarative_base()  # Стара функція

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    is_active = Column(Boolean, default=True)

    # ❌ Немає type hints — IDE не знає тип атрибутів
    # ❌ user.id може бути int або None — незрозуміло без читання Column()

from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):  # Новий стиль — клас, а не виклик функції
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)
    email: Mapped[str] = mapped_column(String(100), unique=True)
    is_active: Mapped[bool] = mapped_column(default=True)

    # ✅ Повні type hints — IDE точно знає типи
    # ✅ Mapped[int] означає NOT NULL, Mapped[int | None] — NULLABLE

Різниця кардинальна. У новому синтаксисі тип стовпця (nullable/not-nullable) тепер виводиться автоматично з типової анотації:

• Mapped[int] → INTEGER NOT NULL
• Mapped[int | None] або Mapped[Optional[int]] → INTEGER NULL
• Mapped[str] → VARCHAR NOT NULL
• Mapped[bool] → BOOLEAN NOT NULL

DeclarativeBase: Базовий клас нового покоління

У SQLAlchemy 2.0 рекомендованим способом створення базового класу для всіх моделей є успадкування від DeclarativeBase замість виклику застарілої функції declarative_base().

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    """
    Базовий клас для всіх ORM-моделей проєкту.
    Всі моделі мають успадковуватися від цього класу.
    """
    pass

Ця проста конструкція робить кілька речей одночасно:

1. Реєструє метадані — кожен підклас Base автоматично реєструє своє визначення таблиці у Base.metadata, що використовується при міграціях Alembic.
2. Забезпечує підтримку type checking — сучасні IDE (PyCharm, VS Code з Pylance) повністю розуміють Mapped[T] і надають автодоповнення та перевірку типів.
3. Дозволяє типізовану конфігурацію через __class_getitem__ механізм — можна визначати type_annotation_map для кастомних типів.

mapped_column(): Детальна конфігурація стовпців

Функція mapped_column() є аналогом старого Column(), але спроєктована для роботи разом із анотаціями Mapped[T]. Вона приймає всі ті самі аргументи, що і Column(), але вже не потребує явного вказання типу (він виводиться з Mapped[T]).

Розглянемо повний приклад моделі User з усіма типовими сценаріями:

from datetime import datetime
from sqlalchemy import String, Text, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    # --- Первинний ключ ---
    id: Mapped[int] = mapped_column(primary_key=True)

    # --- Рядкові поля з обмеженнями ---
    username: Mapped[str] = mapped_column(String(50), unique=True)
    email: Mapped[str] = mapped_column(String(100), unique=True, index=True)

    # --- Nullable поле (опціональне) ---
    # Mapped[str | None] → VARCHAR NULL (поле може бути пустим)
    bio: Mapped[str | None] = mapped_column(Text, default=None)

    # --- Поле з серверним значенням за замовчуванням ---
    # server_default=func.now() → DEFAULT NOW() виконується на стороні PostgreSQL
    created_at: Mapped[datetime] = mapped_column(
        server_default=func.now()
    )

    # --- Поле з клієнтським значенням за замовчуванням ---
    # default=True → значення встановлюється Python-кодом, не SQL
    is_active: Mapped[bool] = mapped_column(default=True)

    def __repr__(self) -> str:
        return f"<User id={self.id} username={self.username!r}>"

Типи стовпців SQLAlchemy

SQLAlchemy надає багатий набір типів стовпців, що відображаються на відповідні SQL-типи конкретної бази даних:

Детальний опис найбільш популярних типів стовпців SQLAlchemy, їхнього призначення, особливостей використання та поширених помилок:

# Приклад використання
id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
views_count: Mapped[int] = mapped_column(Integer, default=0)

# Приклад використання
email: Mapped[str] = mapped_column(String(255), unique=True)
content: Mapped[str] = mapped_column(Text)

# Приклад використання
is_verified: Mapped[bool] = mapped_column(Boolean, default=False)

# Приклад використання
from decimal import Decimal
price: Mapped[Decimal] = mapped_column(Numeric(10, 2))  # Максимум 99999999.99
latitude: Mapped[float] = mapped_column(Float)

# Приклад використання
from datetime import datetime, date
registered_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
birth_date: Mapped[date] = mapped_column(Date)

# Приклад використання
from sqlalchemy.dialects.postgresql import JSONB
metadata_info: Mapped[dict] = mapped_column(JSONB, default=dict)

# Приклад використання
import uuid
from sqlalchemy import Uuid
uuid_id: Mapped[uuid.UUID] = mapped_column(Uuid, default=uuid.uuid4)

# Приклад використання
import enum
from sqlalchemy import Enum

class UserRole(enum.Enum):
    ADMIN = "admin"
    USER = "user"
    GUEST = "guest"

role: Mapped[UserRole] = mapped_column(Enum(UserRole), default=UserRole.USER)

import uuid
from sqlalchemy import Uuid

class Project(Base):
    __tablename__ = "projects"
    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)

Стандартний абстрактний базовий клас з аудитними полями

У реальних проєктах зручно винести загальні поля (наприклад, id, created_at, updated_at) до абстрактного класу, від якого успадковуватимуться всі моделі — так само, як це роблять у EF Core через клас BaseEntity:

public abstract class BaseEntity
{
    public int Id { get; set; }
    public DateTime CreatedAt { get; set; } = DateTime.UtcNow;
    public DateTime? UpdatedAt { get; set; }
}

public class User : BaseEntity
{
    public string Username { get; set; } = default!;
}

from sqlalchemy import String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime

class Base(DeclarativeBase):
    pass

class TimestampMixin:
    """Mixin з аудитними полями для всіх моделей."""
    created_at: Mapped[datetime] = mapped_column(
        server_default=func.now()
    )
    updated_at: Mapped[datetime | None] = mapped_column(
        onupdate=func.now(),
        default=None
    )

class User(TimestampMixin, Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)

Параметр onupdate=func.now() у mapped_column() — це «клієнтський» тригер: SQLAlchemy автоматично підставляє поточний час у UPDATE-запит, якщо об'єкт змінився. Відповідником у PostgreSQL є тригер BEFORE UPDATE, але цей підхід зручніший — логіка знаходиться у Python-коді, а не у SQL.

Relationships: Зв'язки між таблицями

Реляційні бази даних отримали свою назву саме через зв'язки між таблицями. Головна сила ORM полягає в тому, що ці зв'язки можна виразити декларативно у Python-коді, а потім працювати з ними як зі звичайними атрибутами об'єктів.

У SQLAlchemy для визначення зв'язків між моделями використовується функція relationship() у поєднанні з ForeignKey у визначенні стовпця.

Зовнішні ключі: ForeignKey

У той час як relationship() створює зв'язок на рівні Python-об'єктів, клас ForeignKey створює справжнє обмеження цілісності зовнішнього ключа (FOREIGN KEY constraint) безпосередньо у схемі бази даних.

ForeignKey передається як аргумент у mapped_column() для стовпців, що мають посилатися на інші таблиці.

Приклад використання:

# Посилання на таблицю "users" стовпець "id" з каскадним видаленням на рівні СУБД
user_id: Mapped[int] = mapped_column(
    ForeignKey("users.id", ondelete="CASCADE")
)

One-to-Many (Один до Багатьох)

Найпоширеніший тип зв'язку. Один запис у таблиці A може мати декілька пов'язаних записів у таблиці B. Наприклад: один User може мати багато Post-ів.

from __future__ import annotations
from typing import TYPE_CHECKING
from sqlalchemy import String, Text, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)

    # "Один до Багатьох": один User має багато Post-ів
    # back_populates="author" зв'язує цю сторону зі стороною Post.author
    posts: Mapped[list["Post"]] = relationship(
        "Post",
        back_populates="author",
        cascade="all, delete-orphan",  # При видаленні User — видати всі його Post-и
    )

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str | None] = mapped_column(Text)

    # Зовнішній ключ — посилання на users.id
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    # "Зворотний" зв'язок: Post посилається на свого User (автора)
    author: Mapped["User"] = relationship("User", back_populates="posts")

Зверніть на кілька важливих деталей:

1. from __future__ import annotations — це директива, що дозволяє використовувати рядкові посилання на класи ("Post", "User") без циклічного імпорту. Вона змушує Python відкладати обчислення анотацій і є стандартною практикою у файлах із взаємними посиланнями.
2. back_populates — параметр, що встановлює двонаправлений зв'язок: коли ви змінюєте user.posts, SQLAlchemy автоматично оновлює post.author, і навпаки. Це синхронізація в оперативній пам'яті, а не в базі даних.
3. cascade="all, delete-orphan" — правило каскадної поведінки: якщо об'єкт User видаляється, всі його пов'язані Post-и також будуть видалені (аналог ON DELETE CASCADE у SQL, але реалізований на рівні ORM).

Порівняння з EF Core Navigation Properties

У EF Core аналогічна структура виглядала б так:

public class User
{
    public int Id { get; set; }
    public string Username { get; set; } = default!;

    // Navigation Property: колекція пов'язаних Post-ів
    public ICollection<Post> Posts { get; set; } = new List<Post>();
}

public class Post
{
    public int Id { get; set; }
    public string Title { get; set; } = default!;

    // Foreign Key + Navigation Property
    public int AuthorId { get; set; }
    public User Author { get; set; } = default!;
}

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50))

    # Аналог Navigation Property з колекцією
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))

    # Аналог Foreign Key + Navigation Property
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    author: Mapped["User"] = relationship(back_populates="posts")

Many-to-Many (Багато до Багатьох)

Зв'язок «багато до багатьох» реалізується через асоціативну (проміжну) таблицю. Наприклад: один Post може мати багато Tag-ів, і один Tag може належати до багатьох Post-ів.

У SQLAlchemy 2.0 є два підходи для реалізації зв'язку «багато до багатьох».

Конструювання таблиць на рівні Core: Table та Column

Для створення найпростішої проміжної таблиці (без додаткових полів) у SQLAlchemy використовується низькорівневий конструктор Table та клас Column з пакету sqlalchemy (Core-рівень). Оскільки ця таблиця потрібна лише для зв'язку між об'єктами і ми не будемо створювати для неї окремий Python-клас моделі, вона оголошується як об'єкт.

Також у цих таблицях використовуються класи Column (старий Core-еквівалент сучасного ORM-івського mapped_column()):

Підхід 1: Чиста асоціативна таблиця (якщо таблиця не має власних атрибутів):

from sqlalchemy import Table, Column, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

# Проміжна таблиця (без власного класу)
post_tags = Table(
    "post_tags",
    Base.metadata,
    Column("post_id", ForeignKey("posts.id"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id"), primary_key=True),
)

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))

    # Many-to-Many через secondary (проміжну таблицю)
    tags: Mapped[list["Tag"]] = relationship(
        "Tag",
        secondary=post_tags,
        back_populates="posts"
    )

class Tag(Base):
    __tablename__ = "tags"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)

    posts: Mapped[list["Post"]] = relationship(
        "Post",
        secondary=post_tags,
        back_populates="tags"
    )

Підхід 2: Асоціативний клас (якщо проміжна таблиця має власні атрибути, наприклад, role у ProjectMember):

from datetime import datetime
from sqlalchemy import String, ForeignKey, Enum, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
import enum

class Base(DeclarativeBase):
    pass

class MemberRole(enum.Enum):
    OWNER = "owner"
    EDITOR = "editor"
    VIEWER = "viewer"

class ProjectMember(Base):
    """Асоціативна таблиця з власними атрибутами."""
    __tablename__ = "project_members"

    project_id: Mapped[int] = mapped_column(
        ForeignKey("projects.id"), primary_key=True
    )
    user_id: Mapped[int] = mapped_column(
        ForeignKey("users.id"), primary_key=True
    )
    role: Mapped[MemberRole] = mapped_column(
        Enum(MemberRole), default=MemberRole.VIEWER
    )
    joined_at: Mapped[datetime] = mapped_column(server_default=func.now())

    # Navigation properties
    project: Mapped["Project"] = relationship(back_populates="memberships")
    user: Mapped["User"] = relationship(back_populates="memberships")

class Project(Base):
    __tablename__ = "projects"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))

    memberships: Mapped[list["ProjectMember"]] = relationship(
        back_populates="project"
    )

Параметри конфігурації relationship()

Функція relationship() керує об'єднанням моделей на рівні Python. Вона приймає низку критично важливих параметрів, які визначають поведінку зв'язків:

• cascade (тип str, за замовчуванням "save-update, merge") Керує тим, як операції над батьківським об'єктом (наприклад, додавання, оновлення, видалення) поширюються на пов'язані дочірні об'єкти.
  • save-update: якщо додати батьківський об'єкт до сесії (session.add(parent)), усі його дочірні об'єкти автоматично додадуться до сесії.
  • delete: якщо видалити батьківський об'єкт (session.delete(parent)), автоматично видаляться всі пов'язані дочірні об'єкти (аналог ON DELETE CASCADE у БД, але виконується силами Python).
  • delete-orphan (критично важливий): якщо дочірній об'єкт від'єднати від батьківського (наприклад, видалити його зі списку parent.children.remove(child)), цей дочірній об'єкт автоматично буде видалено з БД (інакше він перетвориться на «сироту» — запис у БД з parent_id = NULL).
  • Рекомендований набір: для One-to-Many зв'язків із жорстким володінням (наприклад, User -> Posts) зазвичай ставлять cascade="all, delete-orphan" (all є синонімом для save-update, merge, refresh-expire, expunge, delete).
• passive_deletes (тип bool або "all", за замовчуванням False) Впливає на поведінку каскадного видалення.
  • Якщо False, то при видаленні батьківського об'єкта SQLAlchemy виконає окремий SELECT-запит, щоб завантажити всі дочірні об'єкти в пам'ять, а потім згенерує окремі DELETE-запити для кожного з них. Це створює величезний оверхед.
  • Якщо True, SQLAlchemy НЕ завантажуватиме дочірні об'єкти, а просто видалить батьківський об'єкт. При цьому вона покладатиметься на те, що на рівні самої бази даних у зовнішньому ключі прописано ON DELETE CASCADE (наприклад, ForeignKey("parent.id", ondelete="CASCADE")). Це в рази швидше й ефективніше.
• lazy (тип str, за замовчуванням "select") Визначає стратегію завантаження пов'язаних об'єктів за замовчуванням, якщо її не перевизначено в запиті:
  • "select" (або True): ліниве завантаження (lazy loading).
  • "joined" (або False): жадібне завантаження через LEFT OUTER JOIN (eager loading).
  • "selectin": жадібне завантаження окремим SELECT з IN (eager loading).
  • "raise": кидає помилку при спробі lazy loading. Дуже корисний режим для асинхронного коду, щоб гарантувати, що розробник не забув зробити eager loading і не отримав MissingGreenlet у невідповідний момент.
• back_populates vs backref
  • back_populates (рекомендовано в 2.0) вимагає явного визначення relationship() на обох моделях із взаємним посиланням одна на одну. Це забезпечує повну сумісність із type hints та автодоповненням IDE.
  • backref (застарілий стиль) визначає зв'язок лише на одній моделі, а зворотний зв'язок на іншій створюється динамічно «магічним» шляхом. Це унеможливлює статичну перевірку типів через mypy та призводить до прихованих помилок.

Приклад зв'язку з оптимізованим каскадним видаленням та type safety:

class Project(Base):
    __tablename__ = "projects"
    id: Mapped[int] = mapped_column(primary_key=True)

    # Каскадне видалення дочірніх завдань оптимізовано через passive_deletes
    tasks: Mapped[list["Task"]] = relationship(
        "Task",
        back_populates="project",
        cascade="all, delete-orphan",
        passive_deletes=True,         # SQLAlchemy не буде робити SELECT перед DELETE
    )

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)

    # ON DELETE CASCADE на рівні бази даних для роботи passive_deletes
    project_id: Mapped[int] = mapped_column(
        ForeignKey("projects.id", ondelete="CASCADE")
    )
    project: Mapped["Project"] = relationship(
        "Project",
        back_populates="tasks"
    )

Lazy Loading vs Eager Loading та Проблема N+1

Одна з найбільш підступних проблем у роботі з ORM — це N+1 query problem (проблема N+1 запитів). Вона виникає через поведінку Lazy Loading і є однією з найпоширеніших причин повільної роботи вебдодатків.

Що таке Lazy Loading і чому він небезпечний?

Lazy Loading (ліниве завантаження) — це стратегія, при якій пов'язані об'єкти не завантажуються одразу разом із головним об'єктом. Натомість вони завантажуються «на вимогу» — лише тоді, коли ви вперше звертаєтеся до атрибута зв'язку.

У EF Core Lazy Loading увімкнено за замовчуванням і відбувається автоматично. У SQLAlchemy він також існує, але вимкнений за замовчуванням (особливо у async-режимі, де він взагалі не підтримується без спеціального налаштування).

Уявімо сценарій: отримати список 100 User-ів та для кожного — його username і кількість Post-ів.

З Lazy Loading (Погано ❌):

# 1 запит: SELECT * FROM users LIMIT 100
users = session.execute(select(User).limit(100)).scalars().all()

for user in users:
    # +1 запит для КОЖНОГО користувача:
    # SELECT * FROM posts WHERE author_id = <user.id>
    print(f"{user.username}: {len(user.posts)} posts")

# Разом: 1 + 100 = 101 запит до бази даних!
# При 1000 користувачів — 1001 запит.

Це класична N+1 проблема: спочатку виконується 1 запит для отримання головних об'єктів, а потім для кожного з N об'єктів виконується додатковий запит для отримання пов'язаних даних.

Вирішення: Eager Loading через selectinload() та joinedload()

Eager Loading (жадібне завантаження) вирішує цю проблему, завантажуючи всі необхідні дані одним або двома SQL-запитами. SQLAlchemy надає кілька стратегій:

selectinload() — рекомендований підхід

Виконує два SQL-запити: перший — для головних об'єктів, другий — один SELECT ... WHERE author_id IN (1, 2, 3, ...) для всіх пов'язаних об'єктів одночасно.

from sqlalchemy.orm import selectinload

# Два запити замість N+1:
# 1: SELECT * FROM users LIMIT 100
# 2: SELECT * FROM posts WHERE author_id IN (1, 2, 3, ..., 100)
users = session.execute(
    select(User)
    .options(selectinload(User.posts))
    .limit(100)
).scalars().all()

for user in users:
    print(f"{user.username}: {len(user.posts)} posts")
    # user.posts вже завантажені — жодного додаткового запиту!

joinedload() — для одиничних об'єктів

Виконує один SQL-запит з LEFT OUTER JOIN. Підходить для завантаження одного пов'язаного об'єкта (many-to-one або one-to-one), але може бути неефективним для колекцій:

from sqlalchemy.orm import joinedload

# Один запит з JOIN:
# SELECT posts.*, users.* FROM posts
# LEFT OUTER JOIN users ON users.id = posts.author_id
# WHERE posts.id = 42
post = session.execute(
    select(Post)
    .options(joinedload(Post.author))
    .where(Post.id == 42)
).scalar_one()

print(f"Post by: {post.author.username}")  # Жодного додаткового запиту

Порівняння: .Include() у EF Core ↔ selectinload() у SQLAlchemy

// EF Core: .Include() для завантаження Navigation Property
var users = await context.Users
    .Include(u => u.Posts)  // Аналог selectinload(User.posts)
    .Take(100)
    .ToListAsync();

foreach (var user in users)
{
    Console.WriteLine($"{user.Username}: {user.Posts.Count} posts");
}

# SQLAlchemy: selectinload() для завантаження relationship
users = (await session.execute(
    select(User)
    .options(selectinload(User.posts))
    .limit(100)
)).scalars().all()

for user in users:
    print(f"{user.username}: {len(user.posts)} posts")

Просунуті стратегії завантаження зв'язків

Окрім базових selectinload() та joinedload(), SQLAlchemy надає інструменти для вирішення більш складних архітектурних завдань при завантаженні даних:

• Вкладене завантаження (Nested Eager Loading) Часто потрібно завантажити дерево зв'язків (наприклад, завантажити Project, для нього — всі Task, а для кожного завдання — список Comment та автора коментаря User). Для цього використовується ланцюжок методів selectinload() або joinedload() через крапку:
  nested_loading.py

  from sqlalchemy.orm import selectinload
  
  stmt = (
      select(Project)
      .options(
          selectinload(Project.tasks)           # Крок 1: Завантажити завдання для проекту
          .selectinload(Task.comments)          # Крок 2: Завантажити коментарі для кожного завдання
          .joinedload(Comment.author)           # Крок 3: Жадібно підтягнути автора коментаря (many-to-one)
      )
  )
  projects = (await session.execute(stmt)).scalars().all()
  

• selectinload() vs subqueryload() У старих версіях SQLAlchemy часто використовували subqueryload().
  • subqueryload() повторює вихідний SELECT-запит у підзапиті (subquery) для отримання дочірніх об'єктів. Це може бути вкрай неефективно, якщо вихідний запит мав складні JOIN або великі значення LIMIT/OFFSET (базі даних доводиться виконувати всю логіку вихідного запиту двічі).
  • selectinload() замість цього бере первинні ключі (ID) вже завантажених батьківських об'єктів і робить простий, швидкий запит SELECT ... WHERE parent_id IN (1, 2, 3...). Це набагато легше для планувальника запитів PostgreSQL.
  • Рекомендація: У 2.0 повністю відмовтеся від subqueryload() на користь selectinload().
• contains_eager(): Завантаження зв'язків при ручному JOIN Уявіть ситуацію: вам потрібно відфільтрувати список проектів за статусом їхніх завдань (наприклад, знайти проекти, які мають хоча б одне критичне завдання), і водночас ви хочете, щоб ці завдання були одразу завантажені у властивість project.tasks. Якщо ви напишете .join(Project.tasks).options(selectinload(Project.tasks)), SQLAlchemy зробить два окремих JOIN: один для фільтрації, а другий (в окремому запиті) — для завантаження колекції. Це подвійна робота.
  Правильне рішення — contains_eager(). Воно вказує SQLAlchemy: «Я вже зробив JOIN вручну для фільтрації, візьми ці дані з результату вихідного запиту та поклади у властивість зв'язку»:
  contains_eager_demo.py

  from sqlalchemy.orm import contains_eager
  
  stmt = (
      select(Project)
      .join(Project.tasks)                                   # Ручний JOIN для фільтрації
      .where(Task.priority == TaskPriority.CRITICAL)         # Умова фільтрації
      .options(contains_eager(Project.tasks))                # Завантажити ці ж дані в об'єкт
  )
  # Виконається лише ОДИН SQL-запит із INNER JOIN,
  # і об'єкти project.tasks будуть повністю заповнені!
  projects = (await session.execute(stmt)).scalars().unique().all()
  

Запити у SQLAlchemy 2.0: Core DML API

У SQLAlchemy 1.x робота з запитами на рівні Core (конструктор SQL) та на рівні ORM (робота з об'єктами) була розділена: Core використовував select(users_table), а ORM — session.query(User).

Починаючи з SQLAlchemy 2.0, цей бар'єр повністю стерто. Тепер для будь-яких запитів використовується єдиний уніфікований Core DML (Data Manipulation Language) API. Ми конструюємо запит як об'єкт абстрактного синтаксичного дерева (AST) за допомогою функцій select(), insert(), update() та delete(), а потім виконуємо його через сесію.

Компіляція та інспектування SQL-запитів

Коли ви викликаєте select(User), ви не робите запит до бази даних і навіть не створюєте SQL-рядок. Ви створюєте Python-об'єкт типу Select. SQLAlchemy компілює цей об'єкт у SQL безпосередньо перед виконанням.

Ви можете переглянути згенерований SQL у будь-який момент:

from sqlalchemy import select
from sqlalchemy.dialects import postgresql

stmt = select(User).where(User.username == "arakviel")

# 1. Швидкий перегляд загального SQL (буде використано базовий діалект)
print(stmt)
# Виведе: SELECT users.id, users.username, users.email, users.is_active, users.created_at FROM users WHERE users.username = :username_1

# 2. Перегляд SQL для конкретної СУБД (наприклад, PostgreSQL)
compiled = stmt.compile(dialect=postgresql.dialect())
print(compiled)
# Виведе те саме, але орієнтоване під синтаксис PG

# 3. Перегляд параметрів, які будуть передані з запитом
print(compiled.params)
# Виведе: {'username_1': 'arakviel'}

Завдяки параметризації (використанню плейсхолдерів типу :username_1), SQLAlchemy повністю захищає ваш додаток від SQL-ін'єкцій «з коробки».

Читання даних: select()

Функція select() будує SQL-запити вибірки (SELECT).

Сигнатура

def select(*entities: _ColumnsClauseArgument[Any]) -> Select

Аргументами entities можуть бути класи ORM-моделей (наприклад, User), окремі стовпці (User.username), або SQL-функції (func.count(User.id)).

Ключові методи об'єкта Select (Chaining API)

Кожен метод повертає новий об'єкт Select, що дозволяє будувати ланцюжки викликів:

# Базовий: фільтрація за значенням (рівність або порівняння)
select(User).where(User.is_active == True)
# SQL: SELECT ... FROM users WHERE users.is_active = true

select(User).where(User.id > 10)
# SQL: SELECT ... FROM users WHERE users.id > 10

# Просунутий: комбінація AND, фільтрація NULL, пошук IN та LIKE
select(User).where(
    User.is_active == True,
    User.email.is_not(None),
    User.id.in_([1, 5, 10]),
    User.username.like("ara%")
)
# SQL: SELECT ... FROM users WHERE users.is_active = true AND users.email IS NOT NULL AND users.id IN (1, 5, 10) AND users.username LIKE 'ara%'

# Базовий: фільтрація за одним полем
select(User).filter_by(is_active=True)
# SQL: SELECT ... FROM users WHERE users.is_active = true

# Просунутий: фільтрація за кількома полями одночасно (діє як AND)
select(User).filter_by(is_active=True, username="arakviel")
# SQL: SELECT ... FROM users WHERE users.is_active = true AND users.username = 'arakviel'

# Базовий: сортування за зростанням (за замовчуванням)
select(User).order_by(User.username)
# SQL: SELECT ... FROM users ORDER BY users.username ASC

# Просунутий: сортування за кількома полями, спадання (desc) та NULL в кінці
select(User).order_by(
    User.is_active.desc(),
    User.email.asc().nulls_last()
)
# SQL: SELECT ... FROM users ORDER BY users.is_active DESC, users.email ASC NULLS LAST

# Базовий: обмеження кількості результатів (перші 10 записів)
select(User).limit(10)
# SQL: SELECT ... FROM users LIMIT 10

# Просунутий: ліміт у підзапиті для фільтрації пов'язаних таблиць
subq = select(User.id).order_by(User.created_at.desc()).limit(5).subquery()
select(Post).where(Post.author_id.in_(select(subq)))
# SQL: SELECT ... FROM posts WHERE posts.author_id IN (SELECT users.id FROM users ORDER BY users.created_at DESC LIMIT 5)

# Базовий: пропуск перших 10 записів
select(User).offset(10)
# SQL: SELECT ... FROM users OFFSET 10

# Просунутий: використання разом з .limit() для класичної пагінації
page, limit = 3, 20
select(User).limit(limit).offset((page - 1) * limit)
# SQL: SELECT ... FROM users LIMIT 20 OFFSET 40

# Базовий: групування постів за ID автора
select(Post.author_id, func.count(Post.id)).group_by(Post.author_id)
# SQL: SELECT posts.author_id, count(posts.id) FROM posts GROUP BY posts.author_id

# Просунутий: групування за кількома полями та виразами (екстракція року з дати)
select(
    Post.author_id,
    func.extract("year", Post.created_at),
    func.count(Post.id)
).group_by(
    Post.author_id,
    func.extract("year", Post.created_at)
)
# SQL: SELECT posts.author_id, EXTRACT(year FROM posts.created_at), count(posts.id) FROM posts GROUP BY posts.author_id, EXTRACT(year FROM posts.created_at)

# Базовий: фільтрація груп з кількістю елементів більше 5
select(Post.author_id).group_by(Post.author_id).having(func.count(Post.id) > 5)
# SQL: SELECT posts.author_id FROM posts GROUP BY posts.author_id HAVING count(posts.id) > 5

# Просунутий: фільтрація з перевіркою середньої довжини заголовків постів у групі
select(Post.author_id).group_by(Post.author_id).having(
    func.avg(func.length(Post.title)) > 50
)
# SQL: SELECT posts.author_id FROM posts GROUP BY posts.author_id HAVING avg(length(posts.title)) > 50

# Базовий: INNER JOIN через зв'язок (relationship)
select(User).join(User.posts)
# SQL: SELECT users.id, ... FROM users JOIN posts ON users.id = posts.author_id

# Просунутий: INNER JOIN за явною умовою ON та подвійний JOIN
select(User.username, Tag.name).join(
    Post, User.id == Post.author_id
).join(
    Post.tags
)
# SQL: SELECT users.username, tags.name FROM users JOIN posts ON users.id = posts.author_id JOIN post_tags ON posts.id = post_tags.post_id JOIN tags ON tags.id = post_tags.tag_id

# Базовий: LEFT JOIN користувачів та їхніх постів
select(User, Post).outerjoin(User.posts)
# SQL: SELECT users.id, ... posts.id, ... FROM users LEFT OUTER JOIN posts ON users.id = posts.author_id

# Просунутий: Anti-Join (знайти користувачів, у яких взагалі немає постів)
select(User).outerjoin(User.posts).where(Post.id.is_(None))
# SQL: SELECT users.id, ... FROM users LEFT OUTER JOIN posts ON users.id = posts.author_id WHERE posts.id IS NULL

Оператори та модифікатори стовпців (Column Operators & Modifiers)

Оскільки Python має власні зарезервовані ключові слова (наприклад, in, is, not, and, or), ми не можемо перевантажити їх безпосередньо у виразах на кшталт User.id in [1, 2]. Python-інтерпретатор просто видасть синтаксичну помилку.

Для обходу цього обмеження SQLAlchemy надає спеціальні методи об'єктів стовпців (Column Operators), які генерують відповідні SQL-конструкції:

select(User).where(User.id.in_([1, 2, 3]))
# SQL: SELECT ... FROM users WHERE users.id IN (1, 2, 3)

select(User).where(User.email.is_(None))
# SQL: SELECT ... FROM users WHERE users.email IS NULL

select(User).where(User.email.is_not(None))
# SQL: SELECT ... FROM users WHERE users.email IS NOT NULL

select(User).where(User.username.like("ara%"))
# SQL: SELECT ... FROM users WHERE users.username LIKE 'ara%'

select(User).where(User.username.ilike("%doe%"))
# SQL: SELECT ... FROM users WHERE users.username ILIKE '%doe%'

select(User).where(User.username.startswith("ara"))
# SQL: SELECT ... FROM users WHERE users.username LIKE 'ara%'

select(User).where(User.email.endswith(".com"))
# SQL: SELECT ... FROM users WHERE users.email LIKE '%.com'

select(User).where(User.username.contains("admin"))
# SQL: SELECT ... FROM users WHERE users.username LIKE '%admin%'

select(User).where(User.id.between(10, 20))
# SQL: SELECT ... FROM users WHERE users.id BETWEEN 10 AND 20

select(User).order_by(User.created_at.desc())
# SQL: SELECT ... FROM users ORDER BY users.created_at DESC

select(User).order_by(User.email.asc().nulls_last())
# SQL: SELECT ... FROM users ORDER BY users.email ASC NULLS LAST

select(func.count(User.id).label("total")).scalar_one()
# SQL: SELECT count(users.id) AS total FROM users

from sqlalchemy import cast, String
select(cast(User.id, String))
# SQL: SELECT CAST(users.id AS VARCHAR) FROM users

Як працює «магія» умов (Перевантаження операторів)

Для розробників, які вперше бачать код User.id > 10 у where(), це виглядає як магія. За логікою Python, вираз User.id > 10 мав би негайно обчислитися в логічне True або False (що передало б у where(True) або where(False) і зламало б запит).

Секрет криється в перевантаженні операторів (Operator Overloading) у Python:

1. Стовпці — це не просто значення: Атрибут User.id у вашому коді — це не ціле число int. Це об'єкт класу InstrumentedAttribute (дескриптор, наданий SQLAlchemy).
2. Магічні методи: SQLAlchemy перевизначає стандартні магічні методи порівнянь Python на цих об'єктах:
   • __eq__ (для ==)
   • __ne__ (для !=)
   • __gt__ (для >)
   • __lt__ (для <)
   • тощо.

Коли ви пишете User.id > 10, Python під капотом викликає:

User.id.__gt__(10)

Замість повернення True або False, реалізація __gt__ в SQLAlchemy повертає об'єкт BinaryExpression (бінарний вираз).

Цей об'єкт є вузлом абстрактного синтаксичного дерева (AST) і містить інформацію:

• Ліва частина: посилання на стовпець User.id
• Оператор: > (більше)
• Права частина: константа 10

Коли ви передаєте цей BinaryExpression у метод .where(), SQLAlchemy додає його до списку фільтрів. Під час виконання запиту компилятор діалекту (наприклад, PostgreSQL) обходить це дерево і трансформує вузол у текстовий SQL: users.id > 10 (а точніше, у параметризований вигляд users.id > :id_1).

Зіставлення операторів Python та SQL у SQLAlchemy:

Використання SQL-функцій: func

У багатьох SQL-запитах виникає потреба використовувати вбудовані функції СУБД: порахувати кількість рядків (COUNT), звести текст до нижнього регістру (LOWER), отримати поточний час (NOW()) або виконати складнішу агрегацію.

У SQLAlchemy для цього призначений спеціальний об'єкт func (імпортується як from sqlalchemy import func).

Як працює «динамічна магія» func?

На відміну від більшості бібліотек, у func немає жорстко прописаного списку методів у коді SQLAlchemy. Замість цього він використовує магічний метод Python __getattr__.

Коли ви звертаєтеся до будь-якого атрибута на func (наприклад, func.any_custom_function()), SQLAlchemy «на льоту» створює об'єкт функції з цим ім'ям. Це означає, що ви можете викликати будь-яку функцію, яку підтримує ваша конкретна база даних, навіть якщо розробники SQLAlchemy про неї не знали!

# Будь-яке ім'я динамічно трансформується в SQL-функцію:
stmt = select(func.my_custom_db_function(User.id))
# SQL: SELECT my_custom_db_function(users.id) FROM users

Найпопулярніші стандартні функції СУБД

# Рахуємо загальну кількість користувачів
stmt = select(func.count(User.id))
# SQL: SELECT count(users.id) AS count_1 FROM users

# Отримання поточного часу з серверу БД
stmt = select(func.now())
# SQL: SELECT now() AS now_1

# Якщо bio порожнє (NULL), повернути рядок за замовчуванням
stmt = select(func.coalesce(User.bio, "Біографія відсутня"))
# SQL: SELECT coalesce(users.bio, 'Біографія відсутня') FROM users

# Порівняння email у нижньому регістрі
stmt = select(User).where(func.lower(User.email) == "user@example.com")
# SQL: SELECT ... WHERE lower(users.email) = 'user@example.com'

# Об'єднання імені та прізвища через пробіл
stmt = select(func.concat(User.first_name, " ", User.last_name))
# SQL: SELECT concat(users.first_name, ' ', users.last_name) FROM users

Специфічні функції СУБД (на прикладі PostgreSQL)

Оскільки func динамічний, ви можете викликати специфічні для PostgreSQL функції:

# date_trunc — округлення дати до дня (прибирає години/хвилини)
stmt = select(func.date_trunc("day", User.created_at))
# SQL: SELECT date_trunc('day', users.created_at) FROM users

# age — розрахунок віку на основі дати народження
stmt = select(func.age(User.birth_date))
# SQL: SELECT age(users.birth_date) FROM users

Явне зазначення типу повернення (type_)

Зазвичай SQLAlchemy намагається автоматично визначити тип даних, який поверне функція (наприклад, func.count() поверне ціле число Integer).

Проте для специфічних або кастомних СУБД-функцій SQLAlchemy не знає заздалегідь тип повернення. Через це вона не зможе виконати автоматичну конвертацію результату в об'єкти Python (наприклад, перетворити результат date_trunc у Python-об'єкт datetime.date).

Для вирішення цього використовується параметр type_:

from sqlalchemy import Date

# Явно вказуємо, що функція date_trunc повертає дату
stmt = select(func.date_trunc("day", User.created_at, type_=Date))

# Тепер результат буде автоматично розпарсено у datetime.date в Python:
day = session.execute(stmt).scalar_one()  # поверне datetime.date(2026, 7, 4)

Виконання та отримання результатів

Для виконання запиту об'єкт Select передається в session.execute():

result = session.execute(stmt)

Метод execute() повертає об'єкт Result. Це ітератор, який повертає рядки типу Row (схожі на NamedTuple в Python). Оскільки при виборі повних ORM-моделей кожен рядок повертається як кортеж з одним елементом (наприклад, (User,)), SQLAlchemy надає допоміжні методи для розпакування цих даних:

# Без scalars():
rows = session.execute(select(User)).all()
for row in rows:
    user = row[0]  # доводиться вручну брати перший елемент кортежу
    print(user.username)

# З scalars():
users = session.execute(select(User)).scalars().all()
for user in users:
    print(user.username)  # працюємо безпосередньо з об'єктом User

# 1. Без scalars() — список об'єктів Row ( NamedTuple )
rows = session.execute(select(User.id, User.username)).all()
# rows: list[Row(id=1, username="arakviel"), ...]

# 2. Зі scalars() — список чистих об'єктів User
users = session.execute(select(User)).scalars().all()
# users: list[User]

# Отримання першого знайденого користувача або None
user = session.execute(
    select(User).order_by(User.created_at.asc())
).scalars().first()

# Очікуємо отримати рівно одного користувача за унікальним username
stmt = select(User).where(User.username == "admin")
admin = session.execute(stmt).scalars().one()
# Увага: якщо admin немає — кидає NoResultFound

# Безпечний пошук за унікальним email (поверне об'єкт або None)
stmt = select(User).where(User.email == "notfound@example.com")
user = session.execute(stmt).scalars().one_or_none()

# Отримання кількості користувачів (повертає одне число int)
count = session.execute(
    select(func.count(User.id))
).scalar_one()

# Отримання email користувача за ID (поверне рядок або None, якщо ID не існує)
stmt = select(User.email).where(User.id == 999)
email = session.execute(stmt).scalar_one_or_none()

Практичні приклади з виводом результатів

Приклад 1: Отримання списку активних користувачів (ORM-стиль)

stmt = select(User).where(User.is_active == True).order_by(User.username.asc())

# Компіляція SQL:
# SELECT users.id, users.username, users.email, users.is_active, users.created_at
# FROM users
# WHERE users.is_active = true ORDER BY users.username ASC

users = session.execute(stmt).scalars().all()

for user in users:
    print(f"User: {user.username} (ID: {user.id})")

Вивід у консоль:

User: alice (ID: 2)
User: arakviel (ID: 1)
User: bob (ID: 3)

Приклад 2: Вибірка конкретних стовпців (Tuple-стиль)

Коли нам потрібні лише окремі поля, завантажувати повні ORM-об'єкти неефективно (вони займають пам'ять та відстежуються сесією). Краще вибрати лише потрібні стовпці:

stmt = select(User.id, User.username).where(User.id > 1)

# Компіляція SQL:
# SELECT users.id, users.username FROM users WHERE users.id > 1

rows = session.execute(stmt).all()
# rows містить список об'єктів Row, які поводяться як кортежі та іменовані об'єкти

for row in rows:
    # Доступ можливий двома шляхами:
    print(f"ID: {row[0]}, Username: {row[1]}")          # за індексом
    print(f"ID: {row.id}, Username: {row.username}")    # за атрибутом

Вивід у консоль:

ID: 2, Username: alice
ID: 3, Username: bob

Приклад 3: Складні логічні умови (AND, OR, NOT)

Умови AND поєднуються автоматично при передачі кількох аргументів у where(). Для OR та NOT використовуються спеціальні функції or_ та not_:

from sqlalchemy import or_, not_

stmt = (
    select(User)
    .where(
        User.is_active == True,
        or_(
            User.email.like("%@gmail.com"),
            not_(User.username == "admin")
        )
    )
)

# Компіляція SQL:
# SELECT users.id, users.username, users.email ...
# FROM users
# WHERE users.is_active = true AND (users.email LIKE '%@gmail.com' OR users.username != 'admin')

Приклад 4: INNER JOIN та агрегація (Групування)

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

from sqlalchemy import func

stmt = (
    select(User.username, func.count(Post.id).label("total_posts"))
    .join(Post, User.id == Post.author_id)
    .group_by(User.username)
    .having(func.count(Post.id) > 0)
    .order_by(func.count(Post.id).desc())
)

# Компіляція SQL:
# SELECT users.username, count(posts.id) AS total_posts
# FROM users JOIN posts ON users.id = posts.author_id
# GROUP BY users.username HAVING count(posts.id) > 0
# ORDER BY count(posts.id) DESC

results = session.execute(stmt).all()
for row in results:
    print(f"{row.username}: {row.total_posts} posts")

Вивід у консоль:

arakviel: 12 posts
alice: 3 posts

Створення даних: insert()

Функція insert() генерує SQL-запити вставки записів (INSERT).

Сигнатура

def insert(table: _DMLTableArgument) -> Insert

Аргументом є ORM-клас моделі або об'єкт Table.

Методи об'єкта Insert

• **.values(\*args, **kwargs)** — визначає дані для вставки. Може приймати словник, список словників (для масової вставки) або keyword-аргументи.
• .returning(*cols) — додає секцію RETURNING для отримання згенерованих базою даних значень (наприклад, ID або значень за замовчуванням) в результаті виконання запиту.

ORM vs Core Insertion

У SQLAlchemy є два шляхи створення записів:

Практичні приклади з виводом результатів

Приклад 1: Одиночна вставка з поверненням згенерованого ID

stmt = (
    insert(User)
    .values(username="new_user", email="new@example.com")
    .returning(User.id, User.created_at)
)

# Компіляція SQL (PostgreSQL):
# INSERT INTO users (username, email) VALUES ('new_user', 'new@example.com')
# RETURNING users.id, users.created_at

result = session.execute(stmt)
row = result.fetchone()  # Оскільки ми чекаємо один рядок з RETURNING

print(f"Inserted ID: {row.id}")
print(f"Created At: {row.created_at}")
session.commit()

Вивід у консоль:

Inserted ID: 42
Created At: 2026-07-04 18:00:00+00:00

Приклад 2: Масова вставка (Bulk Insert)

Для оптимізації продуктивності масової вставки передається список словників:

data = [
    {"username": "user_a", "email": "a@example.com"},
    {"username": "user_b", "email": "b@example.com"},
    {"username": "user_c", "email": "c@example.com"},
]

# Створюємо стейтмент без виклику .values() — передамо дані під час виконання
stmt = insert(User)

# Компіляція SQL:
# INSERT INTO users (username, email) VALUES (:username, :email)

result = session.execute(stmt, data)
# SQLAlchemy автоматично групує це в один оптимізований батч-запит

print(f"Rows inserted: {result.rowcount}")
session.commit()

Вивід у консоль:

Rows inserted: 3

Оновлення даних: update()

Функція update() будує SQL-запити оновлення (UPDATE).

Сигнатура

def update(table: _DMLTableArgument) -> Update

Методи об'єкта Update

• .where(*clauses) — обмежує вибірку рядків, які будуть оновлені (критично важливо, інакше оновляться всі рядки в таблиці!).
• .values(**kwargs) або **.values(dict)** — встановлює нові значення для стовпців.
• .returning(*cols) — повертає оновлені значення.

ORM vs Core Update

• ORM-стиль: Отримуємо об'єкт через session.get(User, id), міняємо його атрибут (user.email = "new@mail.com"), і робимо session.commit(). SQLAlchemy автоматично згенерує UPDATE під час комміту завдяки Change Tracking (Unit of Work).
• Core-стиль: Виконуємо update(User).where(...).values(...). Застосовується для масових оновлень або коли потрібно оновити запис без попереднього завантаження його в пам'ять.

Практичні приклади з виводом результатів

Приклад 1: Масове оновлення з фільтрацією

Заблокуємо всіх користувачів, які не заходили на сайт дуже давно (наприклад, у яких пошта на старому домені):

stmt = (
    update(User)
    .where(User.email.like("%@old-domain.com"))
    .values(is_active=False)
)

# Компіляція SQL:
# UPDATE users SET is_active = false WHERE users.email LIKE '%@old-domain.com'

result = session.execute(stmt)
print(f"Matched and updated rows: {result.rowcount}")
session.commit()

Вивід у консоль:

Matched and updated rows: 14

Приклад 2: Динамічне оновлення на основі існуючих значень стовпця

Оновимо лічильник переглядів постів, збільшивши його на 1 directly на рівні бази даних (без завантаження у Python):

stmt = (
    update(Post)
    .where(Post.id == 5)
    .values(views_count=Post.views_count + 1)
    .returning(Post.views_count)
)

# Компіляція SQL:
# UPDATE posts SET views_count = posts.views_count + 1 WHERE posts.id = 5 RETURNING posts.views_count

result = session.execute(stmt)
new_views = result.scalar_one()

print(f"Updated post views count. New value: {new_views}")
session.commit()

Вивід у консоль:

Updated post views count. New value: 101

Видалення даних: delete()

Функція delete() створює SQL-запити видалення записів (DELETE).

Сигнатура

def delete(table: _DMLTableArgument) -> Delete

Методи об'єкта Delete

• .where(*clauses) — визначає умови фільтрації для видалення (якщо не вказати .where(), база даних очистить усю таблицю!).
• .returning(*cols) — повертає дані видалених рядків перед їх остаточним знищенням.

Практичні приклади з виводом результатів

Приклад 1: Видалення неактивних користувачів

stmt = (
    delete(User)
    .where(User.is_active == False)
)

# Компіляція SQL:
# DELETE FROM users WHERE users.is_active = false

result = session.execute(stmt)
print(f"Deleted users count: {result.rowcount}")
session.commit()

Вивід у консоль:

Deleted users count: 5

Приклад 2: Видалення з поверненням видалених даних (RETURNING)

Корисно, якщо перед видаленням нам потрібно залогувати або зберегти видалені дані:

stmt = (
    delete(Post)
    .where(Post.created_at < datetime(2020, 1, 1))
    .returning(Post.id, Post.title)
)

# Компіляція SQL:
# DELETE FROM posts WHERE posts.created_at < '2020-01-01 00:00:00'
# RETURNING posts.id, posts.title

deleted_posts = session.execute(stmt).all()

for post in deleted_posts:
    print(f"Logged deleted post: [ID: {post.id}] {post.title}")

session.commit()

Порівняння: LINQ (EF Core) ↔ SQLAlchemy Query API

// Фільтрація + сортування + пагінація
var users = await context.Users
    .Where(u => u.IsActive && u.Username.StartsWith("ara"))
    .OrderByDescending(u => u.CreatedAt)
    .Skip((page - 1) * perPage)
    .Take(perPage)
    .Include(u => u.Posts)
    .ToListAsync();

// Агрегація
var postCount = await context.Posts
    .Where(p => p.AuthorId == userId)
    .CountAsync();

# Фільтрація + сортування + пагінація
users = (await session.execute(
    select(User)
    .where(User.is_active == True, User.username.like("ara%"))
    .order_by(User.created_at.desc())
    .offset((page - 1) * per_page)
    .limit(per_page)
    .options(selectinload(User.posts))
)).scalars().all()

# Агрегація
from sqlalchemy import func
post_count = (await session.execute(
    select(func.count(Post.id))
    .where(Post.author_id == user_id)
)).scalar_one()

Просунуті операції з даними в Session

Для ефективного написання репозиторіїв та оптимізації роботи з базою даних важливо розрізняти тонкощі поведінки внутрішніх методів Session:

# Масовий імпорт з ігноруванням помилкових записів
for user_data in users_to_import:
    # begin_nested() створює Savepoint і автоматично відкочує його у разі винятку
    async with session.begin_nested():
        try:
            new_user = User(**user_data)
            session.add(new_user)
            await session.flush()  # Спроба вставки конкретного запису
        except Exception as e:
            # Цей запис відкотиться, але сесія залишиться робочою
            print(f"Помилка імпорту {user_data['username']}: {e}")
            continue

await session.commit()  # Зберігаємо всі успішно імпортовані записи

result = await session.execute(select(User, Post).join(User.posts))
for user, post in result:  # Автоматична розпаковка кортежу Row
    print(f"Користувач {user.username} написав пост {post.title}")

Схематичне порівняння методів Session:

Під капотом: Unit of Work, Identity Map та Change Tracking

Щоб дійсно розуміти SQLAlchemy і передбачати її поведінку, недостатньо просто знати синтаксис. Потрібно зрозуміти три ключові патерни, що лежать в основі Session: Unit of Work, Identity Map та Change Tracking.

Патерн Unit of Work (Одиниця роботи)

Unit of Work — це архітектурний патерн (описаний Мартіном Фаулером у книзі «Patterns of Enterprise Application Architecture»), який відстежує всі зміни, що відбуваються з об'єктами під час однієї «одиниці роботи» (зазвичай — одного HTTP-запиту), і потім атомарно відправляє їх до бази даних у вигляді однієї транзакції.

Аналогія: уявіть, що Session — це кошик для покупок. Ви кладете товари (об'єкти) у кошик, берете звідти, змінюєте вміст — і лише коли ви підходите до каси (session.commit()), всі операції виконуються разом як єдина атомарна дія.

sequenceDiagram
    participant Code as "Python Code"
    participant Session as "Session (UoW)"
    participant DB as "PostgreSQL"

    Code->>Session: session.add(new_user)
    Note over Session: Додає об'єкт у стан "pending"
    Code->>Session: user.is_active = False
    Note over Session: Відмічає об'єкт як "dirty"
    Code->>Session: session.delete(old_post)
    Note over Session: Позначає об'єкт як "deleted"
    Code->>Session: await session.commit()
    Session->>DB: BEGIN TRANSACTION
    Session->>DB: INSERT INTO users VALUES (...)
    Session->>DB: UPDATE users SET is_active=false WHERE id=...
    Session->>DB: DELETE FROM posts WHERE id=...
    Session->>DB: COMMIT
    Note over Session: Об'єкти стають "persistent"

Session відстежує об'єкти через чотири можливі стани:

• Transient (перехідний) — об'єкт створений (User()), але ще не доданий до сесії.
• Pending (очікуваний) — об'єкт доданий через session.add(), але ще не збережений до БД (INSERT ще не виконано).
• Persistent (постійний) — об'єкт має відповідний запис у БД та відстежується сесією.
• Detached (від'єднаний) — об'єкт був persistent, але сесія закрита або він був видалений зі сесії.

Identity Map: Гарантія унікальності

Identity Map (карта ідентичностей) — це внутрішній словник у Session, що відображає первинний ключ бази даних на Python-об'єкт.

Ключова гарантія Identity Map: у межах однієї сесії session.get(User, 1) завжди повертає той самий Python-об'єкт, незалежно від кількості викликів.

with Session(engine) as session:
    # Перший виклик: SELECT * FROM users WHERE id = 1
    user_a = session.get(User, 1)

    # Другий виклик: SQL-запиту НЕ виконується!
    # SQLAlchemy знаходить об'єкт у Identity Map і повертає той самий
    user_b = session.get(User, 1)

    print(user_a is user_b)  # True — це буквально один і той самий об'єкт!

    # Якщо ми змінимо user_a — user_b теж «зміниться»
    user_a.is_active = False
    print(user_b.is_active)  # False

Це принципово відрізняється від простого кешування запитів. Identity Map гарантує, що ваша програма завжди працює з консистентним станом об'єктів протягом однієї транзакції.

Порівняння з EF Core: DbContext.ChangeTracker є прямим аналогом Identity Map + Change Tracking у SQLAlchemy. Якщо ви двічі запитуєте один і той самий запис через context.Users.Find(1) в EF Core — ви також отримуєте той самий об'єкт з кешу.

Change Tracking: Як SQLAlchemy «знає», що змінилося

Change Tracking (відстеження змін) — це механізм, завдяки якому Session автоматично знає, які саме атрибути об'єктів були змінені і генерує мінімально необхідний UPDATE-запит.

Цей механізм реалізований через Python-дескриптори (instrumented attributes) — при реєстрації моделі SQLAlchemy замінює звичайні атрибути класу (username, email тощо) на спеціальні об'єкти-дескриптори, що перехоплюють операції __get__ і __set__.

from sqlalchemy import inspect

with Session(engine) as session:
    user = session.get(User, 1)  # Завантажуємо об'єкт

    # Змінюємо кілька атрибутів:
    user.username = "new_username"
    user.bio = "Updated bio"

    # Можна перевірити стан об'єкта ПЕРЕД commit():
    inspector = inspect(user)
    for attr in inspector.attrs:
        history = attr.history
        if history.has_changes():
            print(f"  {attr.key}: {history.deleted} → {history.added}")
    # Виведе:
    # username: ['old_username'] → ['new_username']
    # bio: [None] → ['Updated bio']

    session.commit()
    # SQLAlchemy генерує МІНІМАЛЬНИЙ запит:
    # UPDATE users SET username='new_username', bio='Updated bio' WHERE id=1
    # (лише змінені поля, не всі!)

Саме завдяки Change Tracking SQLAlchemy генерує ефективні UPDATE-запити — оновлюються лише ті стовпці, що дійсно змінилися, а не всі поля об'єкта.

Інтеграція з FastAPI: get_async_session через Dependency Injection

Тепер об'єднаємо все, що ми вивчили, у повноцінний production-ready файл конфігурації бази даних для FastAPI-додатку. Ключовим питанням є: як правильно управляти життєвим циклом AsyncSession — коли її створювати і коли закривати?

Відповідь стандартна для FastAPI: через Dependency Injection (детально розібраний у статті 20).

from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import (
    create_async_engine,
    async_sessionmaker,
    AsyncSession,
)
from sqlalchemy.orm import DeclarativeBase

# 1. Базовий клас для всіх моделей
class Base(DeclarativeBase):
    pass

# 2. Async Engine — singleton, живе весь час роботи додатку
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/taskforge_db"

async_engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_recycle=1800,
    echo=False,
)

# 3. Фабрика AsyncSession
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    autocommit=False,
    autoflush=False,
    expire_on_commit=False,
)

# 4. Dependency для FastAPI — генератор сесії
async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    """
    FastAPI Dependency: надає AsyncSession для одного HTTP-запиту.
    Сесія автоматично закривається після завершення запиту.
    У разі помилки — автоматичний rollback.
    """
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise