SQLAlchemy Shifter

SQLAlchemy Shifter — это легковесный, модульный инструмент для управления миграциями баз данных поверх Python и SQLAlchemy. Идеологически вдохновлен системой миграций фреймворков уровня Yii2 и Laravel.

В отличие от Alembic, здесь нет сложного графа версий, "магической" связки состояний или жесткой привязки к одному файлу alembic.ini. Инструмент использует простой линейный подход, базируясь исключительно на таймстемпах файлов, и спроектирован для использования в современных микросервисных архитектурах (когда миграции могут поставляться разными пакетами).

Ключевые особенности

• Отсутствие Alembic-зависимостей: Работает на чистом SQLAlchemy Core. Встроен собственный минималистичный DDL-генератор, никаких графов версионирования.
• ООП-транзакционность up / safeUp: Базовый класс предоставляет два уровня методов миграции. Если необходимо выполнение внутри транзакции — логика описывается в safeUp(). Движок атомарно обернет её блоком with connection.begin().
• Изоляция через Connection Names: Миграция сама знает, к какой физической базе она относится (через флаг connection = "default"). Это дает возможность управлять несколькими базами данных из одной точки старта.
• Синхронно описываем, Асинхронно исполняем: DDL-миграции пишутся синхронно, а применяются через run_sync в случае использования AsyncEngine.

Установка

Используя Poetry:

poetry add sqlalchemy-shifter

Либо классическим pip:

pip install sqlalchemy-shifter

Использование

1. Создание заготовки миграции

Для создания болванки файла миграции предусмотрен встроенный CLI.

python -m sqlalchemy_shifter.cli create create_users_table --path src/myapp/migrations --conn default

Эта команда сгенерирует файл m20260401_120421_create_users_table.py в указанной папке:

from sqlalchemy_shifter.migration import Migration
from sqlalchemy import Column, Integer, String

class m20260401_120421_create_users_table(Migration):
    
    connection = "default"

    def safeUp(self):
        # Хелперы в стиле Yii2:
        self.create_table(
            "users",
            Column("id", Integer, primary_key=True),
            Column("username", String(255), nullable=False)
        )
        self.create_index("idx_username", "users", ["username"], unique=True)

    def safeDown(self):
        self.drop_index("idx_username", "users")
        self.drop_table("users")

2. Запуск накатки (в коде приложения)

В точке старта вашего приложения инициализируйте ShifterService, передав ему инстансы SQLAlchemy Engines и пути к папкам с реализованными миграциями.

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy_shifter import ShifterService

async def main():
    engine_main = create_async_engine("sqlite+aiosqlite:///app.db")
    
    # 1. Инициализируем сервис
    shifter = ShifterService(
        engines={"default": engine_main},       # Маппинг подключений
        paths=["src/myapp/migrations"]          # Директории с миграциями
    )
    
    # 2. Выполняем миграции
    applied_count = await shifter.upgrade_all()
    print(f"Applied {applied_count} migrations.")

if __name__ == "__main__":
    asyncio.run(main())

Совет: ShifterService.upgrade_all() проверяет таблицу состояния shifter_migrations конкретной базы. Если файл уже в статусе 'applied', он будет проигнорирован.

Лицензия

Пакет поставляется по лицензии MIT. См. файл LICENSE.