alembic_migrations

PostgreSQL + Alembic: миграции в тестах

Интегрируем Alembic в pytest. Фикстура db_connection с автоматическим применением миграций. Session scope для миграций, function scope для rollback. Тестируем на актуальной схеме БД

В production-проектах схема БД постоянно меняется: новые таблицы, колонки, индексы. Alembic — индустриальный стандарт для миграций БД в Python.

Цель: Научиться автоматически применять миграции перед тестами, чтобы тестировать на актуальной схеме.

Prerequisite Check

Вы должны уметь:

# Pytest фикстуры (из Level 1-3)
@pytest.fixture
def db_connection():
    conn = psycopg2.connect(...)
    yield conn
    conn.close()

# Понимать BEGIN/ROLLBACK (из предыдущего урока)
BEGIN;  # Начать транзакцию
# ... SQL команды
ROLLBACK;  # Откатить

Если не уверены — вернитесь к Module 0: Docker PostgreSQL.

Проблема: ручное создание схемы в тестах

Плохой подход (копипаста SQL)

# tests/test_users.py

def test_create_user(db_connection):
    """Тест создания пользователя"""
    cur = db_connection.cursor()

    # ❌ Копипаста CREATE TABLE в каждом тесте
    cur.execute("""
        CREATE TABLE users (
            id SERIAL PRIMARY KEY,
            email VARCHAR(255) NOT NULL UNIQUE
        )
    """)

    cur.execute("INSERT INTO users (email) VALUES (%s)", ("test@test.com",))
    cur.execute("SELECT * FROM users")
    assert cur.fetchone()[1] == "test@test.com"

Проблемы:

❌ Копипаста CREATE TABLE во всех тестах
❌ Схема не совпадает с production (забыли колонку, индекс)
❌ Когда схема меняется — нужно обновить ВСЕ тесты

Чуть лучше: фикстура с SQL

# tests/conftest.py

@pytest.fixture(scope="session")
def db_schema(db_connection):
    """Создаём схему один раз"""
    cur = db_connection.cursor()

    # Загружаем schema.sql
    with open("schema.sql") as f:
        cur.execute(f.read())

    yield

Лучше, но:

❌ schema.sql должен быть синхронизирован с production
❌ Миграции не отслеживаются (какая версия схемы?)
❌ Нет rollback миграций

Решение: Alembic для миграций

Установка Alembic

pip install alembic psycopg2-binary

Инициализация Alembic

# В корне проекта
alembic init alembic

# Создастся структура:
# alembic/
#   env.py           # Конфигурация
#   versions/        # Файлы миграций
# alembic.ini        # Настройки подключения

Настройка alembic.ini

# alembic.ini

[alembic]
script_location = alembic

# ❌ НЕ храните пароли в alembic.ini!
# sqlalchemy.url = postgresql://user:pass@localhost/db

# ✅ Вместо этого настроим в env.py

Настройка alembic/env.py

# alembic/env.py

from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context

# Импортируйте ваши модели (если используете ORM)
# from myapp.models import Base
# target_metadata = Base.metadata

target_metadata = None

config = context.config

# Подключение из переменных окружения
import os
DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "postgresql://testuser:testpass@localhost:5432/testdb"
)
config.set_main_option("sqlalchemy.url", DATABASE_URL)

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

def run_migrations_online():
    """Run migrations in 'online' mode."""
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata
        )

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

run_migrations_online()

Создание первой миграции

# Создайте миграцию для таблицы users
alembic revision -m "create users table"

# Откроется файл: alembic/versions/abc123_create_users_table.py

Заполните миграцию:

# alembic/versions/abc123_create_users_table.py

from alembic import op
import sqlalchemy as sa

# Revision identifiers
revision = 'abc123'
down_revision = None
branch_labels = None
depends_on = None

def upgrade():
    """Применить миграцию"""
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('email', sa.String(length=255), nullable=False),
        sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()')),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('email')
    )

def downgrade():
    """Откатить миграцию"""
    op.drop_table('users')

Применение миграций вручную

# Применить миграции
alembic upgrade head

# Проверить текущую версию
alembic current

# Откатить на одну миграцию назад
alembic downgrade -1

# Откатить все миграции
alembic downgrade base

Интеграция Alembic в pytest

Фикстура: применить миграции перед тестами

# tests/conftest.py

import pytest
import psycopg2
from alembic import command
from alembic.config import Config

@pytest.fixture(scope="session")
def db_migrations():
    """Применить миграции перед тестами (один раз на всю сессию)"""
    # Загрузить Alembic конфиг
    alembic_cfg = Config("alembic.ini")

    # Применить все миграции
    command.upgrade(alembic_cfg, "head")

    yield

    # После тестов откатить (опционально)
    # command.downgrade(alembic_cfg, "base")

@pytest.fixture(scope="session")
def db_engine(db_migrations):
    """Подключение к БД (один раз на сессию)"""
    conn = psycopg2.connect(
        host="localhost",
        port=5432,
        user="testuser",
        password="testpass",
        dbname="testdb"
    )
    yield conn
    conn.close()

@pytest.fixture
def db_connection(db_engine):
    """Транзакция для изоляции теста (на каждый тест)"""
    # Начать транзакцию
    db_engine.execute("BEGIN")

    yield db_engine

    # Откатить после теста (данные исчезнут!)
    db_engine.execute("ROLLBACK")

Как это работает:

db_migrations (session scope): Применяет миграции один раз в начале тестов
db_engine (session scope): Создаёт подключение один раз
db_connection (function scope): Каждый тест получает транзакцию, которая откатывается после теста

✅ Миграции применяются автоматически! ✅ Тесты изолированы (ROLLBACK после каждого)!

Пример теста

# tests/test_users.py

def test_create_user(db_connection):
    """Тест создания пользователя"""
    cur = db_connection.cursor()

    # ✅ Таблица users уже создана миграциями!
    cur.execute("INSERT INTO users (email) VALUES (%s)", ("alice@test.com",))

    cur.execute("SELECT email FROM users WHERE email = %s", ("alice@test.com",))
    result = cur.fetchone()

    assert result[0] == "alice@test.com"

    # ✅ После теста ROLLBACK — данные исчезнут

Запуск тестов

# Применятся миграции → тесты запустятся
pytest tests/test_users.py -v

# Вывод:
# tests/test_users.py::test_create_user PASSED ✅

Практика: добавьте колонку через миграцию

Шаг 1: Создайте миграцию

alembic revision -m "add name column to users"

# Откроется: alembic/versions/def456_add_name_column_to_users.py

Шаг 2: Заполните миграцию

# alembic/versions/def456_add_name_column_to_users.py

def upgrade():
    """Добавить колонку name"""
    op.add_column('users', sa.Column('name', sa.String(length=100), nullable=True))

def downgrade():
    """Удалить колонку name"""
    op.drop_column('users', 'name')

Шаг 3: Тест автоматически использует новую схему

# tests/test_users_with_name.py

def test_create_user_with_name(db_connection):
    """Тест создания пользователя с именем"""
    cur = db_connection.cursor()

    # ✅ Колонка name уже есть (миграция применилась)!
    cur.execute(
        "INSERT INTO users (email, name) VALUES (%s, %s)",
        ("bob@test.com", "Bob")
    )

    cur.execute("SELECT name FROM users WHERE email = %s", ("bob@test.com",))
    result = cur.fetchone()

    assert result[0] == "Bob"

# Миграции автоматически применятся перед тестами
pytest tests/test_users_with_name.py -v

# Вывод:
# tests/test_users_with_name.py::test_create_user_with_name PASSED ✅

Продвинутый паттерн: тестирование миграций

Проблема: миграция может сломать production

Пример опасной миграции:

# ❌ ОПАСНО: удаляем колонку без проверки
def upgrade():
    op.drop_column('users', 'old_email')

Что если в production ещё используется old_email? → Всё сломается!

Решение: тестируйте миграции

# tests/test_migrations.py

import pytest
from alembic import command
from alembic.config import Config

def test_upgrade_downgrade_migrations():
    """Тест что миграции применяются и откатываются без ошибок"""
    alembic_cfg = Config("alembic.ini")

    # Откатить все
    command.downgrade(alembic_cfg, "base")

    # Применить все
    command.upgrade(alembic_cfg, "head")

    # Откатить все обратно
    command.downgrade(alembic_cfg, "base")

    # ✅ Если дошли сюда — миграции работают!

Тестирование конкретной миграции

def test_add_name_column_migration(db_connection):
    """Тест миграции добавления колонки name"""
    cur = db_connection.cursor()

    # Вставьте данные ДО миграции (если миграция уже применена — откатите)
    cur.execute("INSERT INTO users (email) VALUES (%s)", ("test@test.com",))

    # Применить миграцию (если ещё не применена)
    # ...

    # Проверьте что колонка name добавилась
    cur.execute("SELECT name FROM users WHERE email = %s", ("test@test.com",))
    result = cur.fetchone()

    # name должна быть NULL (для старых записей)
    assert result[0] is None

Troubleshooting

"Table 'users' already exists"

Проблема: Миграции уже применены, но Alembic не знает об этом.

Решение: Пометьте текущую версию вручную

# Проверьте текущую версию
alembic current
# (нет версии)

# Пометьте что миграции применены
alembic stamp head

# Теперь Alembic знает о версии
alembic current
# abc123 (head)

"Can't locate revision identified by 'def456'"

Проблема: Alembic не видит новую миграцию.

Решение: Проверьте down_revision

# alembic/versions/def456_add_name_column.py

revision = 'def456'
down_revision = 'abc123'  # ← Должен быть revision предыдущей миграции!

"ROLLBACK не работает — данные остаются"

Проблема: Вы вызываете conn.commit() внутри теста.

Решение: Удалите все commit() из тестов

# ❌ Плохо
def test_create_user(db_connection):
    cur = db_connection.cursor()
    cur.execute("INSERT INTO users (email) VALUES (%s)", ("test@test.com",))
    db_connection.commit()  # ← Удалите это!

# ✅ Хорошо
def test_create_user(db_connection):
    cur = db_connection.cursor()
    cur.execute("INSERT INTO users (email) VALUES (%s)", ("test@test.com",))
    # Данные видны в транзакции, ROLLBACK сработает после теста

Проверьте свои знания

Вопросы ещё не добавлены

Вопросы для этой подтемы ещё не добавлены.

Далее: Фабрики данных: factory_boy + faker