Какой метод есть в миграции для запуска кода Python?

Метод для запуска кода Python в миграциях

В контексте миграций баз данных (Alembic, sqlalchemy-migrate и др.) часто возникает необходимость выполнить Python код наряду с SQL миграциями. Существует несколько подходов в зависимости от инструмента.

1. Alembic — op.execute() и op.run_sync()

op.execute() — выполняет SQL код:

from alembic import op
import sqlalchemy as sa

def upgrade():
    # SQL миграция
    op.create_table('users',
        sa.Column('id', sa.Integer, primary_key=True),
        sa.Column('name', sa.String(50)),
    )
    
    # Выполнить сырой SQL
    op.execute("INSERT INTO users (name) VALUES ('John')")

def downgrade():
    op.drop_table('users')

2. Выполнение Python кода — get_bind()

Для сложной логики используй op.get_bind() для получения соединения с БД:

from alembic import op
import sqlalchemy as sa

def upgrade():
    # Получить соединение
    bind = op.get_bind()
    connection = bind.connection if hasattr(bind, 'connection') else bind
    
    # Выполнить запрос
    result = connection.execute(sa.text("SELECT COUNT(*) FROM users"))
    user_count = result.scalar()
    
    # Python логика
    if user_count > 0:
        print(f"Found {user_count} users")
        connection.execute(sa.text("UPDATE users SET active = true"))

def downgrade():
    pass

3. Выполнение Python функций во время миграции

Для более сложной логики определи функцию и вызови её:

from alembic import op
import sqlalchemy as sa

def process_users(connection):
    """Обработать пользователей и обновить их статусы"""
    result = connection.execute(sa.text("SELECT id, email FROM users"))
    for user_id, email in result:
        # Python логика обработки
        is_valid = validate_email(email)
        status = 'active' if is_valid else 'inactive'
        
        update_sql = "UPDATE users SET status = :status WHERE id = :id"
        connection.execute(
            sa.text(update_sql),
            {"status": status, "id": user_id}
        )
    connection.commit()

def upgrade():
    bind = op.get_bind()
    process_users(bind)

def downgrade():
    pass

def validate_email(email):
    """Простая валидация email"""
    return '@' in email and '.' in email.split('@')[1]

4. Работа с ORM моделями в миграции

Это более сложный, но мощный подход:

from alembic import op
import sqlalchemy as sa
from sqlalchemy.orm import sessionmaker

# Определить простую модель для миграции
Base = sa.orm.declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = sa.Column(sa.Integer, primary_key=True)
    email = sa.Column(sa.String(100), unique=True)
    verified = sa.Column(sa.Boolean, default=False)

def upgrade():
    # Получить сессию
    bind = op.get_bind()
    session = sessionmaker(bind=bind)()
    
    try:
        # Получить всех пользователей
        users = session.query(User).all()
        
        for user in users:
            # Бизнес логика
            if is_email_valid(user.email):
                user.verified = True
        
        session.commit()
    finally:
        session.close()

def downgrade():
    bind = op.get_bind()
    session = sessionmaker(bind=bind)()
    session.query(User).update({User.verified: False})
    session.commit()
    session.close()

def is_email_valid(email):
    return '@' in email

5. Условное выполнение кода по среде

from alembic import op, context
import sqlalchemy as sa

def upgrade():
    # Получить переменные конфига
    config = context.config
    sqlalchemy_url = config.get_main_option("sqlalchemy.url")
    
    # Выполнить код только в production
    if 'postgresql://prod' in sqlalchemy_url:
        bind = op.get_bind()
        bind.execute(sa.text("SELECT * FROM users LIMIT 100"))
    
    op.create_table('new_table',
        sa.Column('id', sa.Integer, primary_key=True),
    )

def downgrade():
    op.drop_table('new_table')

6. Лучшие практики для Python кода в миграциях

Избегай импортов основного приложения:

# ❌ Плохо — может сломаться при развитии приложения
from myapp.models import User

# ✅ Хорошо — дефинируй модели локально
Base = sa.orm.declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = sa.Column(sa.Integer, primary_key=True)

Обработай исключения:

def upgrade():
    bind = op.get_bind()
    try:
        result = bind.execute(sa.text("SELECT COUNT(*) FROM users"))
        count = result.scalar()
        print(f"Processing {count} users")
    except Exception as e:
        print(f"Error: {e}")
        # Решить, откатывать ли миграцию
        raise

Ключевые моменты

• op.execute() — для простого SQL
• op.get_bind() — для получения соединения с БД
• sessionmaker — для работы с ORM и транзакциями
• Дефинируй модели локально — не импортируй из приложения
• Обработай ошибки — миграции должны быть атомарными
• Логируй изменения — помоги себе разобраться в будущем
• Документируй сложную логику — миграции могут быть сложными

Выполнение Python кода в миграциях — мощный инструмент, но используй его осторожно. Лучше полагаться на SQL, когда это возможно, но иногда Python логика необходима для сложных трансформаций данных.