본문으로 건너뛰기
Advertisement

Alembic 마이그레이션

Alembic은 SQLAlchemy 전용 데이터베이스 마이그레이션 도구입니다. Django migrations와 유사하게 스키마 변경 이력을 관리합니다.

설치와 초기화​

pip install alembic

# 프로젝트 초기화
alembic init alembic

# 비동기 환경 초기화
alembic init --template async alembic
myproject/
├── alembic/
│   ├── versions/        # 마이그레이션 파일들
│   ├── env.py           # 환경 설정
│   └── script.py.mako   # 마이그레이션 파일 템플릿
├── alembic.ini          # 주요 설정
└── database.py

설정​

# alembic.ini
[alembic]
script_location = alembic
sqlalchemy.url = postgresql://user:password@localhost/dbname

# 환경 변수 사용 시 (권장)
# sqlalchemy.url = %(DB_URL)s
# alembic/env.py
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context

# 모델 Base 임포트 (자동 감지용)
from database import Base
from models import *  # noqa: F401 — 모든 모델 임포트 필요

config = context.config
fileConfig(config.config_file_name)

target_metadata = Base.metadata


def run_migrations_offline():
    """오프라인 모드: SQL 스크립트 생성"""
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online():
    """온라인 모드: 직접 DB 연결"""
    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,
            compare_type=True,      # 컬럼 타입 변경 감지
            compare_server_default=True,
        )
        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

마이그레이션 명령​

# 현재 모델 기준으로 마이그레이션 자동 생성
alembic revision --autogenerate -m "add product table"

# 빈 마이그레이션 파일 생성 (데이터 마이그레이션용)
alembic revision -m "fill discount rate"

# 최신 버전으로 업그레이드
alembic upgrade head

# 특정 버전으로 업그레이드
alembic upgrade +2         # 현재에서 2단계 앞으로
alembic upgrade abc1234    # 특정 revision ID

# 다운그레이드
alembic downgrade -1       # 1단계 이전으로
alembic downgrade base     # 최초 상태로

# 현재 상태 확인
alembic current

# 이력 보기
alembic history --verbose

마이그레이션 파일 구조​

# alembic/versions/20240115_abc123_add_product_table.py
"""add product table

Revision ID: abc123
Revises: xyz789
Create Date: 2024-01-15 10:00:00
"""
from alembic import op
import sqlalchemy as sa

revision = "abc123"
down_revision = "xyz789"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.create_table(
        "products",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("name", sa.String(200), nullable=False),
        sa.Column("price", sa.Float(), nullable=False),
        sa.Column("stock", sa.Integer(), server_default="0"),
        sa.Column("is_active", sa.Boolean(), server_default="true"),
        sa.Column("category_id", sa.Integer(), sa.ForeignKey("categories.id")),
        sa.Column("created_at", sa.DateTime(), server_default=sa.func.now()),
    )
    op.create_index("idx_products_category", "products", ["category_id", "is_active"])


def downgrade() -> None:
    op.drop_index("idx_products_category")
    op.drop_table("products")

데이터 마이그레이션 패턴​

# alembic/versions/0003_fill_discount_rate.py
from alembic import op
import sqlalchemy as sa
from sqlalchemy.sql import table, column


def upgrade() -> None:
    # 1단계: 컬럼 추가 (null 허용)
    op.add_column(
        "products",
        sa.Column("discount_rate", sa.Float(), nullable=True),
    )

    # 2단계: 데이터 채우기 (SQLAlchemy core 사용)
    products = table("products",
        column("id", sa.Integer),
        column("discount_rate", sa.Float),
    )
    op.execute(
        products.update()
        .where(products.c.discount_rate == None)
        .values(discount_rate=0.0)
    )

    # 3단계: not null 제약 적용
    op.alter_column("products", "discount_rate", nullable=False)


def downgrade() -> None:
    op.drop_column("products", "discount_rate")

비동기 환경 설정​

# alembic/env.py (async 버전)
import asyncio
from sqlalchemy.ext.asyncio import create_async_engine

DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"


def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations():
    engine = create_async_engine(DATABASE_URL)
    async with engine.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await engine.dispose()


def run_migrations_online():
    asyncio.run(run_async_migrations())

실전 팁​

# 마이그레이션 자동 생성 후 반드시 검토!
alembic revision --autogenerate -m "add index"
# → versions/ 폴더의 파일을 열어서 내용 확인 필수

# 스테이징 환경에서 먼저 테스트
DATABASE_URL=postgresql://user:pass@staging/db alembic upgrade head

# SQL 미리보기 (실제 적용 없이)
alembic upgrade head --sql > migration.sql
cat migration.sql  # 검토

정리​

명령설명
alembic init프로젝트 초기화
revision --autogenerate모델 변경 자동 감지
upgrade head최신 마이그레이션 적용
downgrade -1이전 버전으로 롤백
current현재 적용 상태 확인
history전체 이력 보기
RunPython → op.execute데이터 마이그레이션

Alembic의 --autogenerate는 모델과 DB 차이를 자동 감지하지만, 생성된 파일을 배포 전 반드시 검토해야 합니다.

Advertisement
Advertisement