Alembic - 스키마 버전 관리

Alembic이 필요한 이유

DB 스키마를 바꿀 때 흔히 하는 실수가 있다. CREATE TABLE SQL을 직접 실행하거나, 팀원들이 각자 DDL을 돌리는 방식이다.

문제는 재현성이다. 어떤 DDL이 어떤 순서로 실행됐는지 추적이 안 된다. 새 팀원이 환경을 셋업할 때, 스테이징에서 프로덕션으로 배포할 때 매번 수동으로 맞춰야 한다.

Alembic은 마이그레이션을 파일로 관리한다. 각 파일은 upgrade() (스키마 적용)와 downgrade() (롤백)를 담고, 파일 간 연결은 revision ID로 이어진다. alembic upgrade head 한 번으로 현재 DB를 최신 상태로 만들 수 있다.

---

초기 셋업

# 프로젝트 루트에서
alembic init alembic

이 명령어 하나로 alembic/ 디렉토리와 alembic.ini가 생성된다.

alembic/
├── env.py          # 실행 환경 설정 (DB 연결, 메타데이터 등록)
├── script.py.mako  # 마이그레이션 파일 템플릿
└── versions/       # 생성된 마이그레이션 파일들

---

alembic.ini 설정

# alembic.ini
[alembic]
script_location = %(here)s/alembic
prepend_sys_path = .

# DB URL은 여기 하드코딩하지 말고 env.py에서 환경변수로 주입한다
sqlalchemy.url = driver://user:pass@localhost/dbname

sqlalchemy.url의 플레이스홀더는 env.py에서 덮어쓸 거라 크게 신경 안 써도 된다.

---

env.py 설정 — 핵심 부분

자동으로 생성된 env.py를 프로젝트에 맞게 수정해야 한다. 두 가지를 해줘야 한다.

1. DB URL을 환경변수에서 읽기
2. ORM 모델의 메타데이터를 등록해 autogenerate 활성화

# alembic/env.py
import os
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context

# ORM Base를 import해서 메타데이터 등록
from app.database.base import Base
# 모든 엔티티를 import해야 Base.metadata에 테이블 정보가 쌓인다
from app.entity import user, post  # noqa

config = context.config
fileConfig(config.config_file_name)

# autogenerate가 비교 기준으로 사용할 메타데이터
target_metadata = Base.metadata


def get_url() -> str:
    return os.getenv("DATABASE_URL", "postgresql+psycopg2://user:pass@localhost/mydb")


def run_migrations_offline() -> None:
    context.configure(
        url=get_url(),
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    configuration = config.get_section(config.config_ini_section)
    configuration["sqlalchemy.url"] = get_url()
    connectable = engine_from_config(
        configuration,
        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()


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

target_metadata = Base.metadata가 핵심

이게 없으면 Alembic이 "현재 ORM 모델이 어떻게 생겼는지" 모른다. autogenerate 기능(모델 변경을 자동 감지해 마이그레이션 생성)은 이 메타데이터와 실제 DB를 비교해서 동작한다.

주의할 점은 모든 엔티티 파일을 import해야 한다는 것이다. Python은 import되지 않은 모듈의 클래스를 Base.metadata에 등록하지 않기 때문에, 위처럼 # noqa 달고라도 명시적으로 import해줘야 한다.

---

마이그레이션 생성 — autogenerate

# ORM 모델과 실제 DB를 비교해서 마이그레이션 자동 생성
alembic revision --autogenerate -m "create_users_and_posts"

실행하면 alembic/versions/ 안에 파일이 생긴다.

# alembic/versions/abc123_create_users_and_posts.py
"""create_users_and_posts

Revision ID: abc123
Revises:
Create Date: 2025-01-01 12:00:00.000000
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql

revision = 'abc123'
down_revision = None  # 첫 번째 마이그레이션이라 None
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.create_table(
        'users',
        sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
        sa.Column('email', sa.String(length=255), nullable=False),
        sa.Column('username', sa.String(length=50), nullable=False),
        sa.Column('bio', sa.String(length=500), nullable=True),
        sa.Column('is_active', sa.Boolean(), nullable=False),
        sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
        sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('email'),
        sa.UniqueConstraint('username'),
    )
    op.create_table(
        'posts',
        sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
        sa.Column('author_id', postgresql.UUID(as_uuid=True), nullable=False),
        sa.Column('title', sa.String(length=300), nullable=False),
        sa.Column('content', sa.Text(), nullable=True),
        sa.Column('status', sa.Enum('DRAFT', 'PUBLISHED', 'ARCHIVED', name='poststatus'), nullable=False),
        sa.Column('published_at', sa.DateTime(timezone=True), nullable=True),
        sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
        sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False),
        sa.ForeignKeyConstraint(['author_id'], ['users.id']),
        sa.PrimaryKeyConstraint('id'),
    )


def downgrade() -> None:
    op.drop_table('posts')
    op.drop_table('users')
    op.execute("DROP TYPE IF EXISTS poststatus")

upgrade()가 스키마 적용, downgrade()가 롤백이다. autogenerate가 대부분 만들어주지만, Enum 타입 DROP은 직접 추가해줘야 한다. PostgreSQL은 Enum을 별도 타입으로 관리하기 때문에 테이블만 지워도 타입은 남는다.

---

마이그레이션 적용

# 최신 버전까지 적용
alembic upgrade head

# 특정 revision까지만 적용
alembic upgrade abc123

# 한 단계 롤백
alembic downgrade -1

# 현재 DB 상태 확인
alembic current

# 전체 이력 확인
alembic history --verbose

---

실제 워크플로우 — 컬럼 추가 시나리오

posts 테이블에 view_count 컬럼을 추가하는 상황을 가정해보자.

1. ORM 모델에 먼저 추가

# app/entity/post.py
class Post(Base):
    # 기존 컬럼들 ...
    view_count: Mapped[int] = mapped_column(default=0)

2. autogenerate로 마이그레이션 생성

alembic revision --autogenerate -m "add_view_count_to_posts"

생성된 파일:

def upgrade() -> None:
    op.add_column('posts', sa.Column('view_count', sa.Integer(), nullable=False, server_default='0'))

def downgrade() -> None:
    op.drop_column('posts', 'view_count')

3. 적용

alembic upgrade head

코드 변경 → 마이그레이션 생성 → 적용, 이 세 단계가 반복된다.

---

autogenerate가 못 잡는 것들

autogenerate가 편하긴 한데, 모든 변경을 감지하지는 못한다.

• 컬럼 이름 변경: 삭제 + 추가로 감지한다. op.alter_column()으로 직접 써야 한다.
• Enum 값 추가: 테이블 변경이 없어서 감지 안 됨. op.execute("ALTER TYPE ...") 직접.
• PostgreSQL 전용 기능: 파티셔닝, 특수 인덱스 등.

이런 케이스는 alembic revision -m "message"로 빈 파일만 만들고 upgrade()/downgrade()를 직접 작성한다.

---

팀 작업 시 주의점 — 브랜치 충돌

여러 명이 동시에 마이그레이션을 만들면 down_revision이 같은 파일이 여러 개 생긴다. 이걸 브랜치 충돌이라고 부른다.

# 브랜치 확인
alembic branches

# 머지 마이그레이션 생성
alembic merge -m "merge_branches" abc123 def456

머지 파일은 upgrade()/downgrade()가 비어있고, 두 브랜치를 하나로 연결하는 역할만 한다.

실용적인 팁은 PR 올리기 전에 alembic history로 충돌 확인하는 습관이다.

---

정리

Alembic을 쓰면 스키마 변경이 코드처럼 관리된다.

• env.py의 target_metadata 설정이 autogenerate의 핵심
• 모든 엔티티를 import해야 메타데이터에 등록됨
• autogenerate를 믿되, Enum 타입 같은 PostgreSQL 특화 케이스는 직접 확인
• 팀 작업에서 브랜치 충돌은 alembic merge로 해결

다음 글에서는 Prefect 3.x로 파이프라인 전체를 오케스트레이션하는 방법을 다룬다.