Alembic — міграції бази даних

У попередній статті ми навчилися описувати реляційні моделі за допомогою SQLAlchemy 2.0 та виконувати асинхронні операції з базою даних. Ми використовували метод Base.metadata.create_all() для автоматичного створення таблиць при запуску застосунку. Хоча цей підхід виглядає привабливо і просто під час локальної розробки або швидкого прототипування, у реальних виробничих (production) середовищах він є неприпустимим і небезпечним.

У цій статті ми розберемося, як керувати еволюцією схеми бази даних протягом усього життєвого циклу проєкту за допомогою Alembic — офіційного та найбільш потужного інструменту міграцій для SQLAlchemy.

Проблема еволюції схеми БД у Production

Коли застосунок розвивається, його доменна модель неминуче змінюється. Ви додаєте нові сутності, вводите нові колонки, видаляєте застарілі властивості, змінюєте типи даних або створюєте додаткові індекси для оновлення запитів.

Якщо під час розробки на локальному комп'ютері ви можете дозволити собі просто видалити базу даних (виконати DROP DATABASE) та створити її наново з новою схемою, то в реальних системах з реальними користувачами та даними такий підхід призведе до катастрофи.

Чому Base.metadata.create_all() не підходить для Production?

Для розв'язання цих проблем використовуються міграції бази даних — послідовність кроків (скриптів), кожен з яких описує конкретну зміну схеми. Ці скрипти зберігаються в Git і застосовуються послідовно один за одним на кожному середовищі (Development, Staging, Production), гарантуючи, що стан бази даних завжди синхронізований із версією коду застосунку.

Alembic vs Entity Framework Core Migrations

Оскільки ви вже маєте досвід розробки вебзастосунків на платформі .NET та використання Entity Framework Core (EF Core), вам буде легко зрозуміти філософію Alembic через пряме порівняння концепцій та команд.

Хоча екосистема .NET пропонує високоінтегрований CLI для міграцій, Alembic у світі Python надає аналогічні за потужністю можливості, хоча й вимагає дещо більшої уваги до конфігураційних файлів.

Порівняльна таблиця CLI-команд та термінів

Архітектура Alembic та структура проєкту

Alembic — це легковажна та гнучка бібліотека, яка працює безпосередньо поверх драйвера бази даних та SQLAlchemy Engine.

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

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

actor "Developer" as dev

package "Local Workspace (Код)" #f8f9fa {
    component "SQLAlchemy Models\n(Base.metadata)" as models #e3f2fd
    folder "alembic/versions/" as versions_dir #fff3cd {
        file "1a2b3c_init.py\n(Revision 1)" as rev1 #ffe082
        file "4d5e6f_add_users.py\n(Revision 2)" as rev2 #ffe082
    }
    file "env.py" as env_py #e0f2f1
}

package "Target Database (PostgreSQL)" #f8f9fa {
    database "App Schema" as db_schema #d1c4e9 {
        rectangle "users" as tbl_users
        rectangle "projects" as tbl_projects
    }
    rectangle "alembic_version" as db_ver_tbl #ffcdd2 {
        rectangle "version_num : varchar"
    }
}

dev --> models : 1. Модифікує моделі
dev --> env_py : 2. Запускає CLI
env_py --> models : 3. Імпортує метадані
env_py --> versions_dir : 4. Читає/генерує ревізії
env_py --> db_schema : 5. Порівнює та застосовує DDL
db_ver_tbl --> rev2 : 6. Зберігає поточний ID версії ("4d5e6f")

@enduml

Як Alembic відстежує історію: Ланцюжок ревізій (Version Chain)

На відміну від систем міграцій, які покладаються на префікси з таймстемпів у назвах файлів (наприклад, 20260709120000_init.py) для визначення черговості, Alembic використовує концепцію односпрямованого зв'язаного списку (Linked List).

Кожен файл міграції (ревізія) містить унікальний ідентифікатор (revision) та посилання на свого безпосереднього предка (down_revision).

• Перша міграція (Base): не має предка: down_revision = None.
• Наступні міграції: вказують down_revision = 'id_попередньої_міграції'.
• Остання міграція (Head): ревізія, яка знаходиться на вершині списку і на яку наразі немає посилань від інших ревізій.

Завдяки цьому Alembic будує чітке дерево (або лінійний ланцюжок) версій. Якщо ви працюєте у команді й два розробники паралельно створили нові міграції від однієї й тієї самої попередньої версії, Alembic виявить розгалуження (branching) і попросить вас об'єднати їх за допомогою спеціальної команди alembic merge, перш ніж дозволити оновлення бази даних. Це запобігає випадковому затиранню змін при паралельній роботі.

Структура файлів ініціалізованого проєкту

Коли ви запускаєте команду ініціалізації Alembic у вашому проєкті, створюється структура файлів та папок, яка стає ядром системи міграцій:

my_project/
├── alembic.ini             # Глобальний файл конфігурації CLI
└── alembic/                # Директорія середоваща Alembic
    ├── env.py              # Скрипт запуску міграцій
    ├── README              # Документаційний файл
    ├── script.py.mako      # Шаблон для створення нових файлів ревізій
    └── versions/           # Папка з файлами ревізій (міграцій)
        ├── 1a2b3c4d_init.py
        └── 5e6f7g8h_add_users.py

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

1. alembic.ini

Це конфігураційний файл у форматі INI, який зчитується командним інтерфейсом Alembic при кожному запуску. Він містить глобальні параметри, такі як шлях до скрипта env.py, налаштування логування (через стандартний модуль logging у Python), а також рядок підключення до бази даних.

Основне поле, з яким зазвичай взаємодіють розробники на початковому етапі:

sqlalchemy.url = postgresql://user:password@localhost/dbname

!WARNING Зберігання паролів та чутливих даних у файлі alembic.ini у відкритому вигляді є серйозною вразливістю безпеки. Цей файл зазвичай додається до Git, тому жорстке кодування паролів призведе до їх витоку. Далі в статті ми розберемо, як динамічно завантажувати URL бази даних зі змінних середовища за допомогою env.py.

2. alembic/env.py

Це найважливіший файл у структурі Alembic. Це звичайний Python-скрипт, який запускається щоразу, коли ви викликаєте CLI-команду Alembic. Його завдання — налаштувати з'єднання з базою даних (створити SQLAlchemy Engine) та передати це з'єднання об'єкту контексту Alembic для виконання міграцій.

Скрипт env.py має два режими роботи:

• Offline Mode (run_migrations_offline()): Запускається, коли ви викликаєте команду з прапорцем --sql. У цьому режимі Alembic не підключається до бази даних безпосередньо. Замість цього він просто генерує SQL-інструкції (DDL) у текстовий вивід (консоль або файл), які потім DBA (адміністратор бази даних) може застосувати вручну. Це корисно для суворих корпоративних середовищ, де розробники не мають прямого доступу до Production БД.
• Online Mode (run_migrations_online()): Стандартний і найбільш вживаний режим. Alembic створює реальне підключення до бази даних, зчитує поточну версію з таблиці alembic_version, порівнює її з локальними файлами та виконує транзакційне оновлення схеми.

3. alembic/script.py.mako

Це шаблонний файл, написаний на базі шаблонізатора Mako. Коли ви створюєте нову ревізію через команду alembic revision, Alembic використовує цей шаблон для генерації нового Python-файлу у папці versions/. Ви можете кастомізувати цей шаблон, наприклад, додавши до нього імпорти власних типів даних або автогенерацію коментарів для кожного файлу міграції.

4. alembic/versions/

Це сховище ваших міграцій. Кожен файл тут є окремим логічним кроком еволюції вашої бази даних. Ім'я файлу складається з унікального хеш-ідентифікатора ревізії та описового повідомлення, наприклад: 3b82f6a73c12_add_profile_column.py.

Детальний огляд командного інтерфейсу (CLI)

Для ефективної роботи з Alembic вам потрібно опанувати кілька основних команд CLI. Кожна команда має свої параметри та специфіку використання.

Практичний приклад: Налаштування асинхронного Alembic «від А до Я»

Тепер перейдемо від теорії до практики. Ми розберемо життєвий цикл роботи з міграціями на базі невеликого автономного проєкту. Ми створимо просту структуру з двох таблиць (users та posts), ініціалізуємо Alembic за допомогою асинхронного шаблону, розберемося з налаштуванням env.py для асинхронного драйвера aiosqlite (асинхронний SQLite), створимо першу автоматичну міграцію, а потім напишемо ручну міграцію для наповнення бази даних початковими (seed) даними.

Крок 1: Встановлення залежностей та створення моделей

Для початку роботи нам знадобляться SQLAlchemy, Alembic та асинхронний драйвер для SQLite — aiosqlite.

pip install sqlalchemy alembic aiosqlite

uv add sqlalchemy alembic aiosqlite

poetry add sqlalchemy alembic aiosqlite

Створимо файл models.py, де визначимо базовий клас метаданих DeclarativeBase та дві моделі: User та Post з відношенням One-to-Many.

from sqlalchemy import String, 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, nullable=False)
    email: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)
    is_active: Mapped[bool] = mapped_column(default=True)

    # Відношення One-to-Many до постів
    posts: Mapped[list["Post"]] = relationship(
        back_populates="author",
        cascade="all, delete-orphan"
    )

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100), nullable=False)
    content: Mapped[str] = mapped_column(nullable=False)
    author_id: Mapped[int] = mapped_column(
        ForeignKey("users.id", ondelete="CASCADE"),
        nullable=False
    )

    # Зворотне відношення до користувача
    author: Mapped["User"] = relationship(back_populates="posts")

Крок 2: Ініціалізація асинхронного середовища Alembic

Тепер ініціалізуємо середовище Alembic. Оскільки ми розробляємо асинхронний застосунок, стандартна синхронна конфігурація нам не підходить. Нам потрібно викликати alembic init зі спеціальним шаблоном async:

Завдяки параметру -t async Alembic згенерував файл env.py, який налаштований для роботи з асинхронним SQLAlchemy Engine та підтримує запуск міграцій у подієвому циклі (Event Loop).

Крок 3: Конфігурація alembic.ini та env.py

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

Спочатку відкриємо файл alembic.ini та вкажемо шлях до нашої SQLite бази даних. Знайдемо рядок sqlalchemy.url та змінимо його значення:

sqlalchemy.url = sqlite+aiosqlite:///demo.db

Тепер налаштуємо alembic/env.py. Нам потрібно зробити дві речі:

1. Імпортувати об'єкт Base з нашого файлу models.py.
2. Вказати цей об'єкт як target_metadata, щоб Alembic знав, де шукати структуру таблиць.

Відкриємо alembic/env.py та внесемо зміни на початку файлу (на рівні 15-25 рядків):

# 1. Імпортуємо базовий клас моделей
from models import Base

# ... інші імпорти та ініціалізація конфігурації ...

# 2. Передаємо метадані моделей у контекст Alembic
target_metadata = Base.metadata

Глибокий розбір: Як працює асинхронний env.py під капотом?

Якщо ви заглянете всередину згенерованого alembic/env.py, ви побачите суттєві відмінності від класичного синхронного файлу. Давайте розберемо ключові блоки асинхронної конфігурації:

import asyncio
from sqlalchemy.ext.asyncio import async_engine_from_config
from sqlalchemy import pool

def do_run_migrations(connection) -> None:
    """Виконує безпосередній запуск міграцій.
    Ця функція є повністю синхронною.
    """
    context.configure(connection=connection, target_metadata=target_metadata)

    with context.begin_transaction():
        context.run_migrations()

async def run_async_migrations() -> None:
    """Створює асинхронний Engine та асоціює з'єднання з контекстом"""
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    async with connectable.connect() as connection:
        # run_sync запускає синхронну функцію do_run_migrations
        # у контексті асинхронного з'єднання
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()

def run_migrations_online() -> None:
    """Запуск міграцій в режимі 'online'"""
    # Запускаємо асинхронну функцію у новому Event Loop
    asyncio.run(run_async_migrations())

Крок 4: Перша автогенерація міграції (--autogenerate)

Тепер все готово для створення першої міграції. Alembic порівняє стан нашого файлу models.py (де описані таблиці users та posts) з файлом demo.db (якого ще не існує, тобто він порожній) і автоматично згенерує DDL-скрипт для створення таблиць.

Запустимо команду автогенерації:

Зайдемо у папку alembic/versions/ та відкриємо створений файл 7a8b9c10d_initial_schema.py. Он містить приблизно такий код:

"""Initial schema

Revision ID: 7a8b9c10d
Revises:
Create Date: 2026-07-09 11:15:00.000000

"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa

# Ідентифікатори версії
revision: str = '7a8b9c10d'
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None

def upgrade() -> None:
    # 1. Створення таблиці users
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('username', sa.String(length=50), nullable=False),
        sa.Column('email', sa.String(length=100), nullable=False),
        sa.Column('is_active', sa.Boolean(), nullable=False),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('email'),
        sa.UniqueConstraint('username')
    )

    # 2. Створення таблиці posts
    op.create_table(
        'posts',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('title', sa.String(length=100), nullable=False),
        sa.Column('content', sa.String(), nullable=False),
        sa.Column('author_id', sa.Integer(), nullable=False),
        sa.ForeignKeyConstraint(['author_id'], ['users.id'], ondelete='CASCADE'),
        sa.PrimaryKeyConstraint('id')
    )

def downgrade() -> None:
    # Видалення таблиць у зворотному порядку (спочатку posts, потім users)
    op.drop_table('posts')
    op.drop_table('users')

Зверніть увагу, як логічно влаштовані функції:

• upgrade(): застосовує зміни (створює таблиці від головних до залежних).
• downgrade(): повністю скасовує зміни. Важливо, що видалення таблиць відбувається у зворотному порядку (posts видаляється першим), щоб уникнути конфліктів зовнішніх ключів (Foreign Key Constraints).

Анатомія DDL-методів та обмежень (SQLAlchemy & Alembic)

У файлах міграцій ви бачите два основні модулі: alembic.op (операції Alembic) та sqlalchemy (зазвичай імпортований як sa). Давайте розберемо сигнатури, параметри та внутрішній устрій ключових методів та класів, які використовуються для опису схеми бази даних, додавши детальні приклади для кожного з них.

1. Елементи опису структури (SQLAlchemy Core)

sa.Column

Клас sa.Column представляє окреме поле (колонку) у таблиці бази даних. У міграціях його конструктор має наступну сигнатуру:

sa.Column(name: str, type_: TypeEngine, *args, **kwargs)

# Створення простої колонки первинного ключа
sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True)

# Створення обов'язкового рядкового поля зі значенням за замовчуванням
sa.Column('status', sa.String(length=30), nullable=False, server_default='pending')

# Створення часової мітки створення запису (PostgreSQL/SQLite current timestamp)
sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('now()'))

Базові типи даних SQLAlchemy

• sa.Integer(): Відповідає стандартному SQL-типу INTEGER (для цілих чисел).
  Приклад 1: Звичайний цілочисельний IDПриклад 2: Великі числа (BIGINT)2

  sa.Column('user_id', sa.Integer(), nullable=False)
  

• sa.String(length: int | None = None): Відповідає типу VARCHAR(length). Якщо параметр length опущено, СУБД створить необмежений VARCHAR або TEXT (залежно від діалекту, наприклад, у PostgreSQL VARCHAR без довжини є повністю валідним).
  Приклад 1: Обмежений VARCHARПриклад 2: Необмежений VARCHAR (TEXT в Postgres)2

  # Звичайне обмежене поле (наприклад, для номерів телефонів чи кодів)
  sa.Column('phone_number', sa.String(length=20), nullable=True)
  

• sa.Boolean(): Відповідає типу BOOLEAN у PostgreSQL. У СУБД, які не підтримують логічний тип нативно (наприклад, SQLite), він емулюється як TINYINT або INTEGER з Check Constraint (col IN (0, 1)).
  Приклад 1: Активний прапорець зі значенням за замовчуваннямПриклад 2: Логічне поле без значення за замовчуванням2

  sa.Column('is_active', sa.Boolean(), nullable=False, server_default='true')
  

• sa.Text(): Відповідає типу TEXT для збереження великих обсягів символьних даних.
  Приклад 1: Тіло статті чи коментаряПриклад 2: Опціональне поле для великого опису2345678

  sa.Column('body', sa.Text(), nullable=False)
  

::

# Створення легкого опису таблиці користувачів для масового оновлення
users_helper = sa.table('users', sa.column('is_active', sa.Boolean))
op.execute(users_helper.update().values(is_active=True).where(users_helper.c.is_active == False))

# Видалення застарілих логів у системній міграції
logs_helper = sa.table('system_logs', sa.column('created_at', sa.DateTime))
op.execute(logs_helper.delete().where(logs_helper.c.created_at < sa.text("NOW() - INTERVAL '30 days'")))

2. Опис обмежень та зв'язків (Constraints)

sa.PrimaryKeyConstraint

Оголошує первинний ключ таблиці. Може описувати як простий ключ (з однієї колонки), так і складений (композитний) ключ.

sa.PrimaryKeyConstraint(*columns: str, name: str | None = None)

# Звичайний первинний ключ для однієї колонки
sa.PrimaryKeyConstraint('id', name='pk_users')

# Первинний ключ, сформований комбінацією двох колонок (для зв'язку Many-to-Many)
sa.PrimaryKeyConstraint('project_id', 'user_id', name='pk_project_members')

sa.UniqueConstraint

Гарантує, що значення в одній або кількох колонках будуть унікальними для всієї таблиці.

sa.UniqueConstraint(*columns: str, name: str | None = None)

# Просте унікальне обмеження для одного поля
sa.UniqueConstraint('email', name='uq_users_email')

# Запобігає дублюванню однакових тегів у межах одного проекту
sa.UniqueConstraint('project_id', 'slug', name='uq_project_tag_slug')

sa.ForeignKeyConstraint

Описує зв'язок між таблицями (зовнішній ключ). Це один із найважливіших елементів для підтримання цілісності даних (Referential Integrity).

sa.ForeignKeyConstraint(
    columns: list[str],
    refcolumns: list[str],
    onupdate: str | None = None,
    ondelete: str | None = None,
    name: str | None = None
)

# Зв'язок поста з автором: при видаленні користувача його пости видаляються каскадно
sa.ForeignKeyConstraint(['author_id'], ['users.id'], ondelete='CASCADE', name='fk_posts_author_id')

# Зв'язок завдання з виконавцем: при видаленні користувача завдання залишається, але виконавець обнуляється
sa.ForeignKeyConstraint(['assignee_id'], ['users.id'], ondelete='SET NULL', name='fk_tasks_assignee_id')

# Складний зв'язок за двома колонками одночасно
sa.ForeignKeyConstraint(
    ['project_id', 'manager_id'],
    ['projects.id', 'projects.owner_id'],
    ondelete='RESTRICT',
    name='fk_projects_managers'
)

3. Операції Alembic (alembic.op)

Об'єкт op надає набір методів для генерування DDL-інструкцій (зміни структури) та DML-інструкцій (зміни даних).

op.create_table

Емітує SQL-команду CREATE TABLE у базу даних.

op.create_table(table_name: str, *columns_and_constraints, **kwargs)

op.create_table(
    'categories',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('name', sa.String(length=100), nullable=False),
    sa.PrimaryKeyConstraint('id', name='pk_categories')
)

op.create_table(
    'labels',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('name', sa.String(length=50), nullable=False),
    sa.Column('project_id', sa.Integer(), nullable=False),
    sa.ForeignKeyConstraint(['project_id'], ['projects.id'], ondelete='CASCADE', name='fk_labels_project_id'),
    sa.PrimaryKeyConstraint('id', name='pk_labels'),
    sa.UniqueConstraint('project_id', 'name', name='uq_project_label_name')
)

op.drop_table

Емітує SQL-команду DROP TABLE для видалення таблиці та її структури.

op.drop_table(table_name: str, **kwargs)

# Видалення таблиці при відкаті міграції
op.drop_table('categories')

# Зворотний порядок видалення для збереження Referential Integrity у downgrade()
op.drop_table('project_members')
op.drop_table('projects')

op.alter_column

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

op.alter_column(
    table_name: str,
    column_name: str,
    nullable: bool | None = None,
    server_default: Any = None,
    new_column_name: str | None = None,
    type_: TypeEngine | None = None,
    existing_type: TypeEngine | None = None,
    **kwargs
)

# Зміна імені колонки та дозвіл на запис NULL значень
op.alter_column(
    'users',
    'username',
    new_column_name='nickname',
    nullable=True,
    existing_type=sa.String(50)
)

# Зміна типу колонки з Float на BigInteger з вказанням старого типу
op.alter_column(
    'orders',
    'total_amount',
    type_=sa.BigInteger(),
    existing_type=sa.Float()
)

# Додавання системного часу за замовчуванням для існуючої колонки
op.alter_column(
    'tasks',
    'updated_at',
    server_default=sa.text('now()'),
    existing_type=sa.DateTime()
)

op.bulk_insert

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

op.bulk_insert(table: TableClause, rows: list[dict], multi_insert: bool = True)

statuses_tbl = sa.table('task_statuses', sa.column('code', sa.String), sa.column('name', sa.String))
op.bulk_insert(
    statuses_tbl,
    [
        {'code': 'NEW', 'name': 'Нове завдання'},
        {'code': 'IN_PROGRESS', 'name': 'В роботі'},
        {'code': 'COMPLETED', 'name': 'Завершено'}
    ]
)

config_tbl = sa.table('app_configs', sa.column('key', sa.String), sa.column('value', sa.String))
op.bulk_insert(
    config_tbl,
    [
        {'key': 'maintenance_mode', 'value': 'false'},
        {'key': 'max_login_attempts', 'value': '5'}
    ]
)

op.execute

Дозволяє виконати довільний SQL-запит (у вигляді рядка) або об'єкт виразу SQLAlchemy Core. Корисно для складних операцій маніпулювання даними.

op.execute(sql_clause: Any, execution_options: dict | None = None)

# Пряме оновлення даних для міграції існуючих записів
op.execute("UPDATE users SET is_active = true WHERE email LIKE '%@kostyl.dev'")

# Безпечне видалення тестів за допомогою конструктора SQLAlchemy
tasks_tbl = sa.table('tasks', sa.column('title', sa.String))
op.execute(
    tasks_tbl.delete().where(tasks_tbl.c.title.like('TEST:%'))
)

# Встановлення розширення для криптографії, яке не підтримується стандартним Alembic DDL
op.execute("CREATE EXTENSION IF NOT EXISTS pgcrypto")

Анатомія автогенерації: Що Alembic бачить, а що ні?

Функція автогенерації міграцій (--autogenerate) — це надзвичайно зручний помічник, але він не є всемогутнім. Як досвідчений розробник, ви маєте чітко розуміти межі можливостей цього інструменту, щоб не пропустити критичні помилки перед деплоєм на Production.

Крок 5: Застосування міграцій до бази даних

Застосуємо нашу першу міграцію. Це створить файл бази даних demo.db та виконає SQL-запити на створення таблиць.

Тепер перевіримо поточний стан бази даних за допомогою команди alembic current:

Це показує, що наша база даних зараз знаходиться на версії 7a8b9c10d, яка також є вершиною (head) нашого ланцюжка міграцій.

Написання ручних міграцій даних (Data Migrations)

Бувають випадки, коли нам потрібно не просто змінити структуру бази даних (DDL), а модифікувати самі дані (DML). Наприклад, ми хочемо автоматично створити кілька системних записів при розгортанні застосунку.

Давайте створимо порожню міграцію, яку ми заповнимо вручну для додавання початкових (seed) користувачів.

Створимо порожню ревізію:

Оскільки ми не вказали прапорець --autogenerate, Alembic створив файл із порожніми функціями upgrade та downgrade. Давайте відкриємо його та напишемо код для вставки та видалення системних користувачів.

Для вставки даних ми будемо використовувати об'єкт op (Alembic Operations) та його метод bulk_insert(), а для видалення — execute() з конструктором delete від SQLAlchemy Core.

"""seed_initial_users

Revision ID: 9d8e7f6a5
Revises: 7a8b9c10d
Create Date: 2026-07-09 11:30:00.000000

"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa

# Зверніть увагу: тепер down_revision вказує на нашу першу міграцію!
revision: str = '9d8e7f6a5'
down_revision: Union[str, None] = '7a8b9c10d'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None

# Описуємо тимчасову структуру таблиці для безпечної вставки даних
# Це гарна практика, оскільки моделі з models.py можуть змінитися в майбутньому,
# а міграція має залишатися працездатною та ізольованою.
users_table = sa.table(
    'users',
    sa.column('id', sa.Integer),
    sa.column('username', sa.String),
    sa.column('email', sa.String),
    sa.column('is_active', sa.Boolean)
)

def upgrade() -> None:
    # Вставка початкових даних (Seeding)
    op.bulk_insert(
        users_table,
        [
            {
                "id": 1,
                "username": "admin",
                "email": "admin@kostyl.dev",
                "is_active": True
            },
            {
                "id": 2,
                "username": "system_bot",
                "email": "bot@kostyl.dev",
                "is_active": True
            }
        ]
    )

def downgrade() -> None:
    # Видалення доданих даних при відкаті міграції
    op.execute(
        users_table.delete().where(
            users_table.c.id.in_([1, 2])
        )
    )

Застосуємо цю міграцію до бази даних:

Зараз наша база даних містить дві таблиці, один унікальний індекс і два початкових записи. Якщо ми вирішимо відкотити цю міграцію даних, ми можемо виконати команду alembic downgrade -1:

При цьому Alembic виконав наш SQL-запит DELETE і видалив двох користувачів з бази даних, повернувши стан схеми до версії 7a8b9c10d.

Практика: Інтеграція Alembic у наскрізний проєкт TaskForge

Тепер інтегруємо Alembic у наш основний навчальний проєкт TaskForge. На відміну від простого тестового скрипта, у реальному проєкті ми маємо відповідати стандартам Clean Architecture та 12-Factor App. Це означає, що ми не можемо зберігати рядок підключення до бази даних у файлі alembic.ini. Замість цього ми маємо зчитувати його зі змінних середовища та конфігурації проєкту, а також забезпечити автоматичний імпорт усіх наших моделей для коректної роботи автогенерації.

Крок 1: Впровадження Pydantic Settings

У попередній статті (core/database.py) ми жорстко закодували DATABASE_URL. Виправимо це, створивши централізований файл конфігурації app/core/config.py за допомогою бібліотеки pydantic-settings (яку ми вивчили у статті 15).

Створимо файл конфігурації:

from pydantic import PostgresDsn, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore"
    )

    # URL для асинхронного підключення до PostgreSQL
    # Використовуємо тип PostgresDsn для суворої валідації
    DATABASE_URL: str = "postgresql+asyncpg://postgres:postgres@localhost:5432/taskforge"

    @field_validator("DATABASE_URL", mode="before")
    @classmethod
    def assemble_db_connection(cls, v: str | None) -> str:
        if not v:
            raise ValueError("DATABASE_URL must be provided")
        # Переконуємося, що використовується асинхронний драйвер asyncpg
        if v.startswith("postgresql://"):
            return v.replace("postgresql://", "postgresql+asyncpg://", 1)
        return v

settings = Settings()

Тепер оновимо app/core/database.py, щоб він використовував наш синглтон settings:

from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.core.config import settings

# Використовуємо динамічний URL з конфігурації
async_engine = create_async_engine(
    settings.DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_recycle=1800,
)

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

async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

Крок 2: Ініціалізація Alembic у корені проєкту

Перейдемо до кореня нашого проєкту TaskForge та запустимо ініціалізацію асинхронного середовища Alembic.

Крок 3: Налаштування динамічного зчитування URL у env.py

Тепер нам потрібно змінити alembic/env.py, щоб він динамічно зчитував конфігурацію нашого FastAPI-застосунку.

Оскільки скрипт env.py запускається в ізольованому контексті Alembic CLI, за замовчуванням він не бачить модулі нашого застосунку (папку app). Щоб виправити це, додамо шлях до кореня проєкту в sys.path.

Відкриємо alembic/env.py та повністю перепишемо його верхню частину:

import asyncio
import os
import sys
from logging.config import fileConfig

from sqlalchemy import pool
from sqlalchemy.ext.asyncio import async_engine_from_config

from alembic import context

# 1. Додаємо кореневу директорію проєкту до sys.path
# Это дозволяє Alembic імпортувати модулі з пакету 'app'
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))

# 2. Імпортуємо налаштування та базову метамодель застосунку
from app.core.config import settings
from app.models import Base  # Реекспорт усіх моделей

# 3. Налаштування логування
config = context.config
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# 4. Вказуємо метадані для автогенерації
target_metadata = Base.metadata

# 5. Динамічно перезаписуємо sqlalchemy.url з конфігурації Pydantic
config.set_main_option("sqlalchemy.url", settings.DATABASE_URL)

# ... далі йдуть стандартні функції do_run_migrations, run_async_migrations та run_migrations_online ...

Крок 4: Створення початкової автоміграції для PostgreSQL

Переконаємося, що в нашому файлі .env або змінних середовища вказано правильний шлях до нашої PostgreSQL бази даних (наприклад, запущеної у Docker).

Створимо нашу першу автоматичну міграцію, яка опише схему всього проєкту TaskForge:

Глибокий розбір: Як PostgreSQL обробляє Enums?

Якщо ми заглянемо у згенерований файл міграції versions/3a2b1c4d5e6f_initial_schema_specification.py, ми побачимо цікаву деталь, пов'язану з використанням Python-переліків (enum.Enum), таких як MemberRole, TaskStatus та TaskPriority.

На відміну від SQLite (який не має вбудованого типу для енамів і зберігає їх як VARCHAR із Check Constraints), PostgreSQL має повноцінну підтримку користувацьких типів ENUM. Alembic виявляє це та автоматично генерує відповідні SQL-команди:

def upgrade() -> None:
    # 1. Створення PostgreSQL типів ENUM перед створенням таблиць
    member_role_enum = sa.Enum('OWNER', 'EDITOR', 'VIEWER', name='memberrole')
    task_status_enum = sa.Enum('TODO', 'IN_PROGRESS', 'DONE', 'CANCELLED', name='taskstatus')
    task_priority_enum = sa.Enum('LOW', 'MEDIUM', 'HIGH', 'CRITICAL', name='taskpriority')

    # 2. Створення таблиць з використанням цих типів
    op.create_table(
        'project_members',
        sa.Column('project_id', sa.Integer(), nullable=False),
        sa.Column('user_id', sa.Integer(), nullable=False),
        sa.Column('role', member_role_enum, nullable=False),
        sa.Column('joined_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False),
        sa.ForeignKeyConstraint(['project_id'], ['projects.id'], ),
        sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
        sa.PrimaryKeyConstraint('project_id', 'user_id')
    )
    # ...

Це демонструє важливість тестування міграцій саме на тій СУБД, яка буде використовуватися у Production (у нашому випадку — PostgreSQL), оскільки поведінка автогенерації DDL суттєво залежить від діалекту бази даних.

Крок 5: Запуск міграцій на PostgreSQL

Застосуємо створену міграцію до нашої PostgreSQL бази даних:

Тепер наша база даних у PostgreSQL повністю ініціалізована. Вона містить усі необхідні таблиці, зв'язки, індекси, типи ENUM, а також службову таблицю alembic_version із записом 3a2b1c4d5e6f.

Крок 6: Збереження змін у Git

Після успішного локального тестування міграцій обов'язково потрібно зберегти створені файли ревізій та зміни конфігурації у системі контролю версій Git.

Збережемо файли міграцій та налаштування:

Практичні завдання

Для закріплення матеріалу виконайте наступні вправи різного рівня складності.

Базовий рівень

Додайте до проєкту нову модель Category для групування проектів:

1. Опишіть модель Category в окремому файлі app/models/category.py (поля id, name, description).
2. Додайте зв'язок One-to-Many між Category та Project (одна категорія може містити багато проектів).
3. Зареєструйте модель в app/models/__init__.py.
4. Згенеруйте автоматичну міграцію: alembic revision --autogenerate -m "add category model".
5. Перевірте згенерований файл у папці versions/. Переконайтеся, що команди upgrade() та downgrade() описані правильно.
6. Застосуйте міграцію: alembic upgrade head та перевірте структуру таблиць у вашій СУБД.

Середній рівень

Припустимо, ви вирішили додати поле full_name до моделі User, яке має бути обов'язковим (nullable=False):

1. Додайте нову колонку full_name: Mapped[str | None] = mapped_column(String(100), nullable=True) до моделі User.
2. Згенеруйте автоміграцію alembic revision --autogenerate -m "add user full_name".
3. Створіть нову порожню міграцію для перенесення даних: alembic revision -m "populate_user_full_name".
4. У функції upgrade() цієї міграції напишіть SQL-запит (за допомогою об'єкта op та конструктора update), який для кожного користувача скопіює значення з колонки username у колонку full_name (наприклад, як тимчасове ім'я).
5. Тепер змініть поле full_name у моделі User на обов'язкове: nullable=False.
6. Згенеруйте третю автоміграцію alembic revision --autogenerate -m "make full_name non-nullable", яка зробить це поле NOT NULL на рівні схеми бази даних.
7. Застосуйте весь ланцюжок міграцій та переконайтеся, що база даних не видає помилок про порушення NOT NULL обмеження для існуючих записів.

Професійний рівень

Розв'яжіть два складних сценарії, які часто виникають на реальних проєктах:

Сценарій А: Вирішення конфліктів міграцій (Merge Branches)

Уявіть, що ви та ваш колега розробляли паралельно дві різні фічі в окремих Git-гілках:

• Ви додали таблицю tags (ревізія A).
• Колега додав колонку avatar_url до таблиці users (ревізія B).
• Обидві міграції посилаються на одну й ту саму батьківську ревізію Initial schema (ревізія I).

При злитті гілок у main ви отримаєте конфлікт версій в Alembic (два head-елементи: A та B).

1. Спролюйте цю ситуацію локально, скопіювавши файл міграції та зміничив його ідентифікатори так, щоб обидві міграції мали один down_revision.
2. Запустіть alembic heads, щоб побачити список розгалужень.
3. Спробуйте виконати alembic upgrade head та отримайте помилку про наявність кількох голів.
4. Об'єднайте дві голови в одну за допомогою команди:

   alembic merge A B -m "merge tag and avatar migrations"
   

5. Перевірте структуру створеного файлу злиття в versions/ та запустіть оновлення бази даних наново.

Сценарій Б: Програмний запуск міграцій (Lifespan Migration)

У сучасних мікросервісних архітектурах застосунки часто розгортаються в контейнерах Docker (Kubernetes/ECS), де кожна нова версія сервісу має автоматично застосовувати міграції перед стартом самого вебсервера:

1. Замість використання CLI-команд у Bash, налаштуйте автоматичний запуск міграцій Alembic програмно в Event Loop вашого FastAPI застосунку.
2. Використовуйте подію lifespan у app/main.py:

   import os
   from contextlib import asynccontextmanager
   from fastapi import FastAPI
   from alembic.config import Config
   from alembic import command
   
   def run_migrations():
       alembic_cfg = Config("alembic.ini")
       command.upgrade(alembic_cfg, "head")
   
   @asynccontextmanager
   async def lifespan(app: FastAPI):
       # Запуск міграцій при старті застосунку в синхронному потоці
       # щоб не блокувати головний Event Loop
       import anyio
       await anyio.to_thread.run_sync(run_migrations)
       yield
   
   app = FastAPI(lifespan=lifespan)
   

3. Протестуйте це рішення, видаливши базу даних та запустивши застосунок за допомогою uvicorn main:app. Переконайтеся, що при старті сервера таблиці були створені автоматично.