Panduan Lengkap Menggunakan Alembic: Database Migration Made Easy

Alembic adalah tool migration database untuk SQLAlchemy yang powerful dan fleksibel. Tool ini memungkinkan Anda melacak perubahan skema database secara version control, membuat migration scripts, dan menjalankan upgrade atau downgrade database dengan mudah.

Dalam tutorial ini, kita akan mempelajari Alembic dari dasar hingga advanced usage dengan contoh praktis.

Mengapa Alembic?

Keunggulan Alembic:

Version Control untuk Database: Melacak setiap perubahan skema database

Reversible Migrations: Mendukung upgrade dan downgrade

Autogenerate: Otomatis generate migration dari perubahan model SQLAlchemy

Branching: Mendukung multiple migration branches

SQLAlchemy Integration: Terintegrasi sempurna dengan SQLAlchemy ORM

Offline Mode: Generate SQL scripts tanpa koneksi database

Instalasi

Install Alembic

# Install Alembic pip install alembic Install dengan SQLAlchemy (jika belum) pip install sqlalchemy Install database driver sesuai kebutuhan pip install psycopg2-binary # PostgreSQL pip install pymysql # MySQL pip install aiosqlite # SQLite async

Verifikasi Instalasi

alembic --version Output: alembic 1.13.x

Setup Project: Konfigurasi Awal

1. Inisialisasi Alembic

# Buat folder project mkdir myproject && cd myproject Inisialisasi Alembic alembic init alembic

Struktur folder yang dihasilkan:

myproject/
├── alembic/
│   ├── versions/           # Folder untuk migration files
│   ├── env.py              # Environment configuration
│   ├── README              # Dokumentasi
│   └── script.py.mako      # Template untuk migration
├── alembic.ini             # Konfigurasi utama Alembic

2. Konfigurasi Database URL

Edit file alembic.ini:

# alembic.ini
[alembic]
scriptlocation = alembic
prependsyspath = .

Database URL
sqlalchemy.url = postgresql://user:password@localhost/mydatabase

Untuk SQLite
sqlalchemy.url = sqlite:///./app.db

Untuk MySQL
sqlalchemy.url = mysql+pymysql://user:password@localhost/mydatabase

3. Setup Model SQLAlchemy

Buat file models.py:

# models.py
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Boolean, Text
from sqlalchemy.orm import declarativebase, relationship

from datetime import datetime

Base = declarativebase()

class User(Base):
    tablename = 'users'

    id = Column(Integer, primarykey=True)

    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    passwordhash = Column(String(255), nullable=False)
    isactive = Column(Boolean, default=True)

    createdat = Column(DateTime, default=datetime.utcnow)

    # Relationship
    posts = relationship("Post", backpopulates="author")


class Post(Base):
    tablename = 'posts'

    id = Column(Integer, primarykey=True)
    title = Column(String(200), nullable=False)
    content = Column(Text)
    userid = Column(Integer, ForeignKey('users.id'), nullable=False)

    createdat = Column(DateTime, default=datetime.utcnow)
    updatedat = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)


    # Relationship
    author = relationship("User", backpopulates="posts")

4. Konfigurasi env.py untuk Autogenerate

Edit file alembic/env.py:

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

Import Base dari models
import sys
from pathlib import Path
sys.path.append(str(Path(file).parent.parent))
from models import Base

Alembic Config object
config = context.config

Setup logging
if config.configfilename is not None:
    fileConfig(config.configfilename)

Set targetmetadata untuk autogenerate

targetmetadata = Base.metadata

def runmigrationsoffline() -> None:
    """Run migrations in 'offline' mode."""
    url = config.getmainoption("sqlalchemy.url")
    context.configure(
        url=url,
        targetmetadata=targetmetadata,
        literalbinds=True,

        dialectopts={"paramstyle": "named"},
    )

    with context.begintransaction():

        context.runmigrations()

def runmigrationsonline() -> None:
    """Run migrations in 'online' mode."""
    connectable = enginefromconfig(
        config.getsection(config.configinisection, {}),

        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            targetmetadata=targetmetadata

        )

        with context.begintransaction():
            context.runmigrations()


if context.isofflinemode():

    runmigrationsoffline()

else:
    runmigrationsonline()

Membuat Migration

1. Manual Migration

# Buat migration kosong alembic revision -m "create users table"

File migration yang dihasilkan (alembic/versions/xxxxcreateuserstable.py):

"""create users table

Revision ID: a1b2c3d4e5f6
Revises:
Create Date: 2024-01-15 10:30:00.000000

"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa

revision identifiers
revision: str = 'a1b2c3d4e5f6'
downrevision: Union[str, None] = None

branchlabels: Union[str, Sequence[str], None] = None
dependson: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    op.createtable(
        'users',
        sa.Column('id', sa.Integer(), primarykey=True),

        sa.Column('username', sa.String(50), unique=True, nullable=False),
        sa.Column('email', sa.String(100), unique=True, nullable=False),
        sa.Column('passwordhash', sa.String(255), nullable=False),
        sa.Column('isactive', sa.Boolean(), default=True),

        sa.Column('createdat', sa.DateTime(), default=sa.func.now()),
    )

def downgrade() -> None:
    op.droptable('users')

2. Autogenerate Migration

Alembic dapat otomatis mendeteksi perubahan pada model SQLAlchemy:

# Generate migration dari perubahan model alembic revision --autogenerate -m "add posts table"

Alembic akan membandingkan:

Model SQLAlchemy saat ini (dari targetmetadata)
Skema database yang ada

Dan menghasilkan migration yang mencakup:

Tabel baru
Kolom baru
Perubahan tipe data
Index dan constraint baru

3. Contoh Autogenerate Output

"""add posts table

Revision ID: b2c3d4e5f6g7
Revises: a1b2c3d4e5f6
Create Date: 2024-01-15 11:00:00.000000

"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa

revision: str = 'b2c3d4e5f6g7'
downrevision: Union[str, None] = 'a1b2c3d4e5f6'

branchlabels: Union[str, Sequence[str], None] = None
dependson: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    op.createtable(
        'posts',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('title', sa.String(length=200), nullable=False),
        sa.Column('content', sa.Text(), nullable=True),
        sa.Column('userid', sa.Integer(), nullable=False),

        sa.Column('createdat', sa.DateTime(), nullable=True),
        sa.Column('updatedat', sa.DateTime(), nullable=True),

        sa.ForeignKeyConstraint(['userid'], ['users.id'], ),
        sa.PrimaryKeyConstraint('id')
    )

def downgrade() -> None:
    op.droptable('posts')

Menjalankan Migration

1. Upgrade Database

# Upgrade ke revision terbaru alembic upgrade head Upgrade ke revision tertentu alembic upgrade a1b2c3d4e5f6 Upgrade satu step alembic upgrade +1 Upgrade dua step alembic upgrade +2

2. Downgrade Database

# Downgrade satu step alembic downgrade -1 Downgrade ke revision tertentu alembic downgrade a1b2c3d4e5f6 Downgrade ke awal (hapus semua) alembic downgrade base

3. Melihat Status Migration

# Lihat revision saat ini alembic current Lihat history semua migration alembic history Lihat history dengan detail alembic history --verbose Lihat migration yang belum dijalankan alembic heads

Output contoh alembic history:

b2c3d4e5f6g7 -> head (add posts table) a1b2c3d4e5f6 -> b2c3d4e5f6g7 (create users table) -> a1b2c3d4e5f6 (empty message)

Operasi Migration Umum

1. Menambah Kolom

from alembic import op
import sqlalchemy as sa

def upgrade() -> None:
    op.addcolumn('users', sa.Column('phone', sa.String(20), nullable=True))
    op.addcolumn('users', sa.Column('avatarurl', sa.String(255), nullable=True))

def downgrade() -> None:
    op.dropcolumn('users', 'avatarurl')
    op.dropcolumn('users', 'phone')

2. Mengubah Kolom

from alembic import op
import sqlalchemy as sa

def upgrade() -> None:
    # Ubah tipe data
    op.altercolumn('users', 'username',
                    existingtype=sa.String(50),

                    type=sa.String(100))

    # Ubah nullable
    op.altercolumn('posts', 'content',

                    existingtype=sa.Text(),
                    nullable=False)

def downgrade() -> None:
    op.altercolumn('users', 'username',

                    existingtype=sa.String(100),
                    type=sa.String(50))


    op.altercolumn('posts', 'content',
                    existingtype=sa.Text(),

                    nullable=True)

3. Menghapus Kolom

def upgrade() -> None:
    op.dropcolumn('users', 'phone')

def downgrade() -> None:
    op.addcolumn('users', sa.Column('phone', sa.String(20), nullable=True))

4. Membuat Index

def upgrade() -> None:
    # Single column index
    op.createindex('ixusersemail', 'users', ['email'])

    # Composite index
    op.createindex('ixpostsusercreated', 'posts', ['userid', 'createdat'])

    # Unique index
    op.createindex('ixusersusername', 'users', ['username'], unique=True)


def downgrade() -> None:
    op.dropindex('ixusersusername', 'users')
    op.dropindex('ixpostsusercreated', 'posts')
    op.dropindex('ixusersemail', 'users')

5. Membuat Foreign Key

def upgrade() -> None:
    op.createforeignkey(

        'fkpostsuserid',      # constraint name
        'posts',                  # source table
        'users',                  # referent table
        ['userid'],              # local columns

        ['id'],                   # remote columns
        ondelete='CASCADE'
    )

def downgrade() -> None:
    op.dropconstraint('fkpostsuserid', 'posts', type='foreignkey')

6. Data Migration

from alembic import op
import sqlalchemy as sa
from sqlalchemy.sql import table, column

def upgrade() -> None:
    # Define table untuk operasi data
    users = table('users',
        column('id', sa.Integer),
        column('username', sa.String),
        column('role', sa.String)
    )

    # Insert data
    op.bulkinsert(users, [

        {'username': 'admin', 'role': 'admin'},
        {'username': 'user1', 'role': 'user'},
    ])

    # Update data
    op.execute(
        users.update()
        .where(users.c.username == 'admin')
        .values(role='superadmin')
    )

def downgrade() -> None:
    # Delete data
    users = table('users', column('username', sa.String))
    op.execute(users.delete().where(users.c.username.in(['admin', 'user1'])))

Branching dan Merging

1. Membuat Branch

# Buat branch baru dari revision tertentu alembic revision -m "feature branch" --head a1b2c3d4e5f6 Lihat semua heads (branches) alembic heads

2. Merging Branches

Ketika ada multiple branches yang perlu digabung:

# Buat merge migration alembic merge -m "merge feature branches" branch1head branch2head

"""merge feature branches

Revision ID: merge123456
Revises: branch1head, branch2head
Create Date: 2024-01-15 14:00:00.000000

"""
revision = 'merge123456'
downrevision = ('branch1head', 'branch2head')


def upgrade():
    pass

def downgrade():
    pass

Offline Mode: Generate SQL

1. Generate SQL Script

# Generate SQL tanpa eksekusi alembic upgrade head --sql > migration.sql Generate SQL untuk range tertentu alembic upgrade a1b2c3d4e5f6:head --sql > migration.sql

2. Review SQL Script

-- migration.sql
BEGIN;

CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50) NOT NULL UNIQUE,
    email VARCHAR(100) NOT NULL UNIQUE,
    passwordhash VARCHAR(255) NOT NULL,
    isactive BOOLEAN DEFAULT TRUE,

    createdat TIMESTAMP DEFAULT CURRENTTIMESTAMP

);

INSERT INTO alembicversion (versionnum) VALUES ('a1b2c3d4e5f6');


COMMIT;

Integrasi dengan FastAPI

1. Struktur Project

myfastapiapp/
├── alembic/
│   ├── versions/
│   ├── env.py
│   └── script.py.mako
├── app/
│   ├── init.py
│   ├── main.py
│   ├── database.py
│   ├── models/
│   │   ├── init.py
│   │   └── user.py
│   └── schemas/
│       └── user.py
├── alembic.ini
└── requirements.txt

2. Database Configuration

# app/database.py
from sqlalchemy import createengine
from sqlalchemy.orm import sessionmaker, declarativebase


DATABASEURL = "postgresql://user:password@localhost/mydatabase"

engine = createengine(DATABASEURL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarativebase()


def getdb():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

3. Model dengan Base Terpusat

# app/models/user.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from datetime import datetime
from app.database import Base

class User(Base):
    tablename = 'users'

    id = Column(Integer, primarykey=True, index=True)

    email = Column(String, unique=True, index=True)
    username = Column(String, unique=True, index=True)
    hashedpassword = Column(String)
    isactive = Column(Boolean, default=True)

    createdat = Column(DateTime, default=datetime.utcnow)

4. env.py untuk FastAPI

# alembic/env.py
import sys
from pathlib import Path

Add app to path
sys.path.append(str(Path(file).parent.parent))

from app.database import Base
from app.models import user  # Import semua models

targetmetadata = Base.metadata

Environment Variables untuk Database URL

1. Menggunakan Environment Variable

Edit alembic/env.py:

import os
from dotenv import loaddotenv

loaddotenv()


def geturl():
    return os.getenv("DATABASEURL", "sqlite:///./app.db")


def runmigrationsoffline() -> None:

    url = geturl()
    context.configure(
        url=url,
        targetmetadata=targetmetadata,
        literalbinds=True,

    )
    with context.begintransaction():
        context.runmigrations()


def runmigrationsonline() -> None:

    configuration = config.getsection(config.configinisection)
    configuration["sqlalchemy.url"] = geturl()


    connectable = enginefromconfig(

        configuration,
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    # ... rest of the function

2. File .env

# .envDATABASEURL=postgresql://user:password@localhost/mydatabase

Best Practices

1. Naming Convention yang Konsisten

# Format: actiontabledescription
alembic revision -m "createuserstable"
alembic revision -m "addemailcolumntousers"
alembic revision -m "createindexonusersemail"
alembic revision -m "addforeignkeypoststousers"

2. Selalu Test Migration

# Test upgrade dan downgrade
def testmigration():
    # Upgrade
    alembiccfg = Config("alembic.ini")

    command.upgrade(alembiccfg, "head")

    # Downgrade
    command.downgrade(alembiccfg, "base")


    # Upgrade lagi untuk memastikan reversible
    command.upgrade(alembiccfg, "head")

3. Backup Sebelum Migration di Production

# PostgreSQL
pgdump mydatabase > backupbeforemigration.sql


MySQL
mysqldump mydatabase > backupbeforemigration.sql


Jalankan migration
alembic upgrade head

4. Review Autogenerate Output

Autogenerate tidak selalu sempurna. Selalu review migration yang di-generate:

# Hal yang TIDAK terdeteksi autogenerate:
- Rename table/column (terdeteksi sebagai drop + create)
- Perubahan pada constraint name
- Perubahan pada data

Selalu edit manual jika perlu rename
def upgrade() -> None:
    # Autogenerate: drop oldcolumn, add newcolumn

    # Manual fix: rename column
    op.altercolumn('users', 'oldcolumn', newcolumnname='newcolumn')

5. Atomic Migrations

Satu migration = satu perubahan logis:

# Good: Satu migration per fitur
alembic revision -m "adduserprofiletable"

alembic revision -m "addavatarcolumntousers"


Bad: Terlalu banyak perubahan dalam satu migration
alembic revision -m "addeverythingforuserfeature"

6. Downgrade yang Aman

def downgrade() -> None:
    # Selalu pastikan downgrade bisa dijalankan
    # Perhatikan data yang mungkin hilang

    # Jika ada data penting, backup dulu
    conn = op.getbind()
    conn.execute("CREATE TABLE usersbackup AS SELECT  FROM users")


    # Baru drop
    op.droptable('users')

Troubleshooting

1. Multiple Heads Error

# Error: Multiple heads found alembic heads # Lihat semua heads Solution: Merge heads alembic merge -m "merge heads" head1 head2

2. Target Database is Not Up to Date

# Error: Can't locate revision alembic stamp head # Set current revision tanpa menjalankan migration

3. Revision Not Found

# Jika revision file hilang tapi database sudah diupdate alembic stamp base # Reset ke base alembic upgrade head # Jalankan semua migration

4. Autogenerate Tidak Mendeteksi Perubahan

# Pastikan import semua models di env.py
from app.models import user, post, comment  # Import explicit

Atau import dengan wildcard
from app.models import

Kesimpulan

Alembic adalah tool yang sangat powerful untuk database migration:

Key Takeaways:

Version Control Database: Setiap perubahan skema terlacak

Reversible: Mendukung upgrade dan downgrade

Autogenerate: Otomatis detect perubahan model SQLAlchemy

Flexible: Mendukung berbagai database (PostgreSQL, MySQL, SQLite, dll)

Production Ready: Offline mode untuk generate SQL scripts

Kapan Menggunakan Alembic:

Use Alembic when:

Menggunakan SQLAlchemy sebagai ORM
Butuh version control untuk database schema
Tim development dengan multiple developers
Aplikasi dengan frequent schema changes
Production deployment yang memerlukan rollback capability

Don't use Alembic when:

Simple scripts yang tidak memerlukan database
Database non-relational (gunakan tools khusus MongoDB, Redis, dll)
One-time data import tanpa schema changes

Command Cheatsheet:

# Setup alembic init alembic Create migration alembic revision -m "description" alembic revision --autogenerate -m "description" Run migration alembic upgrade head alembic downgrade -1 Status alembic current alembic history Generate SQL alembic upgrade head --sql

Resources:

Alembic Docs: https://alembic.sqlalchemy.org
SQLAlchemy Docs: https://docs.sqlalchemy.org
GitHub: https://github.com/sqlalchemy/alembic

Selamat menggunakan Alembic!