Complete Guide to Using Alembic: Database Migration Made Easy

Alembic is a powerful and flexible database migration tool for SQLAlchemy. It enables you to track database schema changes with version control, create migration scripts, and easily perform database upgrades or downgrades.

In this tutorial, we'll learn Alembic from basics to advanced usage with practical examples.

Why Alembic?

Alembic Advantages:

Version Control for Database: Track every database schema change

Reversible Migrations: Support for upgrade and downgrade

Autogenerate: Automatically generate migrations from SQLAlchemy model changes

Branching: Support for multiple migration branches

SQLAlchemy Integration: Seamless integration with SQLAlchemy ORM

Offline Mode: Generate SQL scripts without database connection

Installation

Install Alembic

# Install Alembic pip install alembic Install with SQLAlchemy (if not already installed) pip install sqlalchemy Install database driver as needed pip install psycopg2-binary # PostgreSQL pip install pymysql # MySQL pip install aiosqlite # SQLite async

Verify Installation

alembic --version Output: alembic 1.13.x

Project Setup: Initial Configuration

1. Initialize Alembic

# Create project folder mkdir myproject && cd myproject Initialize Alembic alembic init alembic

Generated folder structure:

myproject/
├── alembic/
│   ├── versions/           # Folder for migration files
│   ├── env.py              # Environment configuration
│   ├── README              # Documentation
│   └── script.py.mako      # Template for migrations
├── alembic.ini             # Main Alembic configuration

2. Configure Database URL

Edit alembic.ini file:

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

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

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

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

3. Setup SQLAlchemy Models

Create models.py file:

# 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. Configure env.py for Autogenerate

Edit alembic/env.py file:

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

Import Base from 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 for 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()

Creating Migrations

1. Manual Migration

# Create empty migration alembic revision -m "create users table"

Generated migration file (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 can automatically detect changes in SQLAlchemy models:

# Generate migration from model changes alembic revision --autogenerate -m "add posts table"

Alembic compares:

Current SQLAlchemy models (from targetmetadata)
Existing database schema

And generates migrations that include:

New tables
New columns
Data type changes
New indexes and constraints

3. Example 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')

Running Migrations

1. Upgrade Database

# Upgrade to latest revision alembic upgrade head Upgrade to specific revision alembic upgrade a1b2c3d4e5f6 Upgrade one step alembic upgrade +1 Upgrade two steps alembic upgrade +2

2. Downgrade Database

# Downgrade one step alembic downgrade -1 Downgrade to specific revision alembic downgrade a1b2c3d4e5f6 Downgrade to beginning (remove all) alembic downgrade base

3. Check Migration Status

# View current revision alembic current View history of all migrations alembic history View history with details alembic history --verbose View pending migrations alembic heads

Example alembic history output:

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

Common Migration Operations

1. Adding Columns

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. Altering Columns

from alembic import op
import sqlalchemy as sa

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

                    type=sa.String(100))

    # Change 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. Dropping Columns

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

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

4. Creating Indexes

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. Creating Foreign Keys

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 for data operations
    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 and Merging

1. Creating a Branch

# Create new branch from specific revision alembic revision -m "feature branch" --head a1b2c3d4e5f6 View all heads (branches) alembic heads

2. Merging Branches

When multiple branches need to be merged:

# Create 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 without execution alembic upgrade head --sql > migration.sql Generate SQL for specific range 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;

Integration with FastAPI

1. Project Structure

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 with Centralized Base

# 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 for 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 all models

targetmetadata = Base.metadata

Environment Variables for Database URL

1. Using Environment Variables

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. .env File

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

Best Practices

1. Consistent Naming Convention

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

2. Always Test Migrations

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

    command.upgrade(alembiccfg, "head")

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


    # Upgrade again to ensure reversibility
    command.upgrade(alembiccfg, "head")

3. Backup Before Production Migration

# PostgreSQL
pgdump mydatabase > backupbeforemigration.sql


MySQL
mysqldump mydatabase > backupbeforemigration.sql


Run migration
alembic upgrade head

4. Review Autogenerate Output

Autogenerate isn't always perfect. Always review generated migrations:

# Things NOT detected by autogenerate:
- Rename table/column (detected as drop + create)
- Changes to constraint names
- Data changes

Always edit manually for renames
def upgrade() -> None:
    # Autogenerate: drop oldcolumn, add newcolumn

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

5. Atomic Migrations

One migration = one logical change:

# Good: One migration per feature
alembic revision -m "adduserprofiletable"

alembic revision -m "addavatarcolumntousers"


Bad: Too many changes in one migration
alembic revision -m "addeverythingforuserfeature"

6. Safe Downgrades

def downgrade() -> None:
    # Always ensure downgrade can be executed
    # Be aware of data that might be lost

    # If there's important data, backup first
    conn = op.getbind()
    conn.execute("CREATE TABLE usersbackup AS SELECT  FROM users")


    # Then drop
    op.droptable('users')

Troubleshooting

1. Multiple Heads Error

# Error: Multiple heads found alembic heads # View all 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 without running migration

3. Revision Not Found

# If revision file is missing but database is already updated alembic stamp base # Reset to base alembic upgrade head # Run all migrations

4. Autogenerate Not Detecting Changes

# Make sure to import all models in env.py
from app.models import user, post, comment  # Explicit import

Or import with wildcard
from app.models import

Conclusion

Alembic is a powerful tool for database migration:

Key Takeaways:

Database Version Control: Every schema change is tracked

Reversible: Supports upgrade and downgrade

Autogenerate: Automatically detects SQLAlchemy model changes

Flexible: Supports various databases (PostgreSQL, MySQL, SQLite, etc.)

Production Ready: Offline mode for generating SQL scripts

When to Use Alembic:

Use Alembic when:

Using SQLAlchemy as ORM
Need version control for database schema
Development team with multiple developers
Applications with frequent schema changes
Production deployment requiring rollback capability

Don't use Alembic when:

Simple scripts that don't require a database
Non-relational databases (use specific tools for MongoDB, Redis, etc.)
One-time data import without 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

Happy coding with Alembic!