SQLAlchemy — Alembic Migrations
Setup
uv add alembic
alembic init alembic # sync template
alembic init -t async alembic # async template (choose one, not both)
Configure alembic/env.py
from app.database import Base # your DeclarativeBase
from app.config import settings
config = context.config
config.set_main_option("sqlalchemy.url", settings.database_url)
target_metadata = Base.metadata
Set target_metadata to your Base.metadata so Alembic can detect model changes.
Core Commands
| Command | Purpose |
|---|---|
alembic revision --autogenerate -m "add users table" |
Create migration from model diff |
alembic upgrade head |
Apply all pending migrations |
alembic downgrade -1 |
Roll back one migration |
alembic downgrade base |
Roll back all migrations |
alembic current |
Show current revision |
alembic history |
List all revisions |
alembic heads |
Show latest revision(s) |
alembic stamp head |
Mark DB as up-to-date without running migrations |
Auto-Generated Migration
alembic revision --autogenerate -m "add users table"
Generates a file like alembic/versions/abc123_add_users_table.py:
"""add users table"""
from alembic import op
import sqlalchemy as sa
revision = "abc123"
down_revision = None
def upgrade() -> None:
op.create_table(
"users",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("name", sa.String(100), nullable=False),
sa.Column("email", sa.String(255), nullable=False),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("email"),
)
op.create_index("ix_users_email", "users", ["email"])
def downgrade() -> None:
op.drop_index("ix_users_email", "users")
op.drop_table("users")
Always review auto-generated migrations — they may miss renames, data migrations, or ordering.
Common Operations
Add Column
def upgrade() -> None:
op.add_column("users", sa.Column("bio", sa.Text(), nullable=True))
def downgrade() -> None:
op.drop_column("users", "bio")
Rename Column
def upgrade() -> None:
op.alter_column("users", "name", new_column_name="full_name")
def downgrade() -> None:
op.alter_column("users", "full_name", new_column_name="name")
Add Index
def upgrade() -> None:
op.create_index("ix_users_name", "users", ["name"])
def downgrade() -> None:
op.drop_index("ix_users_name", "users")
Add Foreign Key
def upgrade() -> None:
op.add_column("posts", sa.Column("user_id", sa.Integer(), nullable=False))
op.create_foreign_key("fk_posts_user_id", "posts", "users", ["user_id"], ["id"])
def downgrade() -> None:
op.drop_constraint("fk_posts_user_id", "posts", type_="foreignkey")
op.drop_column("posts", "user_id")
Data Migrations
from alembic import op
import sqlalchemy as sa
from sqlalchemy import text
def upgrade() -> None:
op.add_column("users", sa.Column("role", sa.String(20), server_default="viewer"))
op.execute(text("UPDATE users SET role = 'admin' WHERE is_superuser = true"))
op.drop_column("users", "is_superuser")
def downgrade() -> None:
op.add_column("users", sa.Column("is_superuser", sa.Boolean(), server_default="false"))
op.execute(text("UPDATE users SET is_superuser = true WHERE role = 'admin'"))
op.drop_column("users", "role")
Async Migrations (env.py)
import asyncio
from alembic import context
from sqlalchemy.ext.asyncio import create_async_engine
from app.database import Base
from app.config import settings
def do_migrations(connection):
context.configure(connection=connection, target_metadata=Base.metadata)
with context.begin_transaction():
context.run_migrations()
async def run_async_migrations():
engine = create_async_engine(settings.database_url)
async with engine.connect() as connection:
await connection.run_sync(do_migrations)
await engine.dispose()
def run_migrations_online():
asyncio.run(run_async_migrations())
run_migrations_online()
Best Practices
- Always review autogenerate output — check renames, data integrity, index order.
- Write both
upgrade()anddowngrade()— rollbacks save production incidents. - Use
server_defaultfor new NOT NULL columns — avoids failures on existing rows. - Test migrations on a copy of production data — catches edge cases early.
- Run
alembic upgrade headin CI — verify migrations apply cleanly. - Squash old migrations periodically — keeps history manageable.
- Never edit already-applied migrations — create a new revision instead.