Можно ли передать односвязный список в функцию len в Python?

Привязка Python кода к миграциям БД

Да, это не только возможно, но иногда необходимо. Но нужно быть очень аккуратным, чтобы не создать хрупкую систему. Разберём когда и как это делать.

Когда Python код в миграциях полезен

1. Data migration (миграция данных)

Когда вы меняете структуру данных, может потребоваться Python код для трансформации.

-- migrations/0003_migrate_user_names.sql
-- Тут нужно разбить full_name на first_name и last_name

ALTER TABLE users ADD COLUMN first_name VARCHAR(100);
ALTER TABLE users ADD COLUMN last_name VARCHAR(100);

-- Но КАК разбить "Ivan Petrov" на first_name и last_name?
-- Нужен Python код!

2. Исправление невалидных данных

Если в БД накопились невалидные данные (NULL в required поле, дублирующиеся email'ы), Python позволит их исправить.

3. Заполнение вычисляемых полей

Если добавляете новое поле, которое зависит от других данных, Python может вычислить значения.

Django migrations (Alembic, который используется в Django)

В Django встроена поддержка Python code в миграциях:

# migrations/0004_populate_user_slugs.py
from django.db import migrations
from django.utils.text import slugify

def populate_slugs(apps, schema_editor):
    User = apps.get_model('auth', 'User')
    for user in User.objects.all():
        user.slug = slugify(user.username)
        user.save(update_fields=['slug'])

def reverse_populate_slugs(apps, schema_editor):
    User = apps.get_model('auth', 'User')
    User.objects.all().update(slug=None)

class Migration(migrations.Migration):
    dependencies = [
        ('auth', '0003_alter_user_add_slug'),
    ]
    
    operations = [
        migrations.RunPython(populate_slugs, reverse_populate_slugs),
    ]

Это безопасно потому что:

• Миграция имеет функцию reverse (откат)
• Работает с моделями через apps.get_model(), а не прямо импортирует
• Версионируется и отслеживается в контроле версий

SQLAlchemy Alembic

Если вы используете Alembic (как в современных FastAPI приложениях), там тоже можно выполнять Python код:

# alembic/versions/0004_populate_computed_field.py
from alembic import op
import sqlalchemy as sa
from datetime import datetime

def upgrade():
    # 1. Сначала добавляем столбец
    op.add_column('orders', sa.Column('computed_total', sa.Float, nullable=True))
    
    # 2. Заполняем данные через Python
    connection = op.get_bind()
    
    # Получаем все заказы
    result = connection.execute(
        sa.text("SELECT id, items FROM orders WHERE computed_total IS NULL")
    )
    
    for row in result:
        # Вычисляем total
        total = calculate_total(row.items)  # Python функция
        
        connection.execute(
            sa.text("UPDATE orders SET computed_total = :total WHERE id = :id"),
            {"total": total, "id": row.id}
        )

def downgrade():
    op.drop_column('orders', 'computed_total')

Goose (raw SQL) — специальный подход

Если вы используете Goose (чистый SQL), то Python код в миграциях требует специального подхода. Есть несколько вариантов:

Вариант 1: SQL процедуры вместо Python

Напишите логику миграции через PL/pgSQL (для PostgreSQL):

-- migrations/0005_migrate_names.sql
-- Используем PL/pgSQL вместо Python

DO $$
DECLARE
    r RECORD;
BEGIN
    FOR r IN SELECT id, full_name FROM users WHERE first_name IS NULL
    LOOP
        UPDATE users SET
            first_name = split_part(r.full_name, ' ', 1),
            last_name = split_part(r.full_name, ' ', 2)
        WHERE id = r.id;
    END LOOP;
END
$$;

Вариант 2: Отдельный Python скрипт

Используйте Python скрипт ПОСЛЕ применения миграции:

#!/bin/bash
# scripts/post_migration_0005.sh

echo "Applying migration..."
goose -dir migrations postgres "$DATABASE_URL" up

echo "Running Python data migration..."
python scripts/migrate_user_names.py

# scripts/migrate_user_names.py
import os
from sqlalchemy import create_engine, text

engine = create_engine(os.getenv('DATABASE_URL'))

with engine.connect() as conn:
    # Получаем всех пользователей
    result = conn.execute(
        text("SELECT id, full_name FROM users WHERE first_name IS NULL")
    )
    
    for row in result:
        parts = row.full_name.split(' ', 1)
        first_name = parts[0]
        last_name = parts[1] if len(parts) > 1 else ''
        
        conn.execute(
            text("""
            UPDATE users SET first_name = :fn, last_name = :ln
            WHERE id = :id
            """),
            {"fn": first_name, "ln": last_name, "id": row.id}
        )
    
    conn.commit()

print("Migration completed!")

Вариант 3: Миграция в два этапа

Сначала добавляете столбец с DEFAULT, потом обновляете данные отдельно:

-- migrations/0006_add_slug.sql
ALTER TABLE users ADD COLUMN slug VARCHAR(255);

# Потом в коде приложения (например, в management command)
from django.core.management.base import BaseCommand
from django.utils.text import slugify

class Command(BaseCommand):
    def handle(self, *args, **options):
        from users.models import User
        
        # Обновляем данные
        for user in User.objects.filter(slug__isnull=True):
            user.slug = slugify(user.username)
            user.save()
        
        self.stdout.write(self.style.SUCCESS('Slugs populated'))

Запускаете после развертывания:

goose up
python manage.py populate_slugs  # После успешного deploy

Best practices

1. Всегда делайте миграции идемпотентными

# ✅ ХОРОШО: Проверяем, не выполнена ли уже миграция
def upgrade():
    connection = op.get_bind()
    
    # Проверяем, есть ли уже обновлённые данные
    result = connection.execute(
        sa.text("SELECT COUNT(*) FROM users WHERE computed_total IS NULL")
    )
    
    if result.scalar() > 0:
        # Только тогда обновляем
        # ... код миграции

2. Сохраняйте старые модели в миграции

# ❌ ПЛОХО: Использовать текущую модель
from users.models import User  # Может измениться в будущем!

def upgrade():
    User.objects.all().update(...)  # Опасно!

# ✅ ХОРОШО: Использовать историческую модель
def upgrade(apps, schema_editor):
    User = apps.get_model('users', 'User')  # Снимок модели на момент миграции

3. Для больших таблиц используйте батчи

def upgrade(apps, schema_editor):
    connection = op.get_bind()
    
    # Обновляем порциями, не разом всю таблицу
    batch_size = 1000
    offset = 0
    
    while True:
        rows = connection.execute(
            sa.text(
                "SELECT id FROM users LIMIT :limit OFFSET :offset"
            ),
            {"limit": batch_size, "offset": offset}
        )
        
        rows_list = list(rows)
        if not rows_list:
            break
        
        # Обновляем этот батч
        for row in rows_list:
            # ... код обновления
            pass
        
        offset += batch_size

4. Тестируйте миграции

# Тестируйте откат
make migrate
make migrate-down
make migrate-up

# Проверьте, что данные целые
psql -c "SELECT COUNT(*) FROM users WHERE slug IS NULL;"

Итог

Da, Python код в миграциях полезен, но лучше использовать специализированные инструменты:

• Django: RunPython операции встроены
• Alembic: Python код выполняется в migration файлах
• Goose: Используйте SQL процедуры или отдельные скрипты

Главное правило: миграции должны быть идемпотентными, обратимыми и протестированными.