Writing Alembic Migrations by Hand

While autogenerate handles routine schema changes, some migrations are too complex or require data transformations. Writing migrations by hand gives you full control and is essential for renaming columns, splitting columns, computing new values, or running custom SQL. Alembic provides the op object—a database-agnostic builder API—that lets you express any schema or data change in Python.

The op Object: Your Migration Toolkit

The op object is imported in every migration file and provides methods to alter your schema and run arbitrary SQL. Common methods:

• op.add_column(table, column) — add a new column
• op.drop_column(table, column) — remove a column
• op.alter_column(table, column, ...) — modify a column (type, nullable, rename)
• op.create_table(name, columns) — create a new table
• op.drop_table(name) — drop a table
• op.create_index(name, table, columns) — add an index
• op.execute(sql) — run arbitrary SQL; also supports Python expressions
• op.create_foreign_key(...) — add a foreign key constraint

Creating a Manual Migration

Generate an empty migration file:

alembic revision -m "Split name into first and last"

This creates a file like 003_split_name_into_first_and_last.py with empty upgrade() and downgrade() functions. Open it and write your logic.

Example 1: Rename a Column

If you renamed user_name to full_name in your model, autogenerate produces a drop-and-add (which loses data). Write it correctly:

# migrations/versions/003_rename_user_name_to_full_name.py
def upgrade():
    op.alter_column('users', 'user_name', new_column_name='full_name')

def downgrade():
    op.alter_column('users', 'full_name', new_column_name='user_name')

The new_column_name parameter renames the column in place, preserving all data.

Example 2: Split a Column with Data Transformation

You have a name column and want to split it into first_name and last_name. This requires:

1. Add the new columns
2. Compute values from the old column
3. Drop the old column

# migrations/versions/004_split_name_column.py
def upgrade():
    # Step 1: Add new columns (nullable while we populate them)
    op.add_column('users', sa.Column('first_name', sa.String(50)))
    op.add_column('users', sa.Column('last_name', sa.String(50)))
    
    # Step 2: Execute a SQL statement to split the name
    op.execute(
        """
        UPDATE users
        SET first_name = SUBSTRING_INDEX(name, ' ', 1),
            last_name = SUBSTRING_INDEX(name, ' ', -1)
        WHERE name IS NOT NULL
        """
    )
    
    # Step 3: Make new columns non-nullable and drop the old one
    op.alter_column('users', 'first_name', nullable=False, existing_type=sa.String(50))
    op.alter_column('users', 'last_name', nullable=False, existing_type=sa.String(50))
    op.drop_column('users', 'name')

def downgrade():
    # Reverse: add name column, concatenate first/last, drop new columns
    op.add_column('users', sa.Column('name', sa.String(100)))
    op.execute(
        """
        UPDATE users
        SET name = CONCAT(first_name, ' ', last_name)
        """
    )
    op.drop_column('users', 'first_name')
    op.drop_column('users', 'last_name')

Note: SQL syntax varies by database (MySQL uses SUBSTRING_INDEX, PostgreSQL uses split_part). For database-agnostic migrations, see below.

Example 3: Conditional Logic with Python

If you need to compute values based on Python logic (e.g., calculate derived fields), use op.execute() with a context connection:

# migrations/versions/005_add_user_status.py
from sqlalchemy import text, select, and_

def upgrade():
    op.add_column('users', sa.Column('status', sa.String(20), nullable=True))
    
    # Get a database connection to fetch and update rows
    connection = op.get_bind()
    
    # Example: set status based on created_at date
    op.execute(
        text("""
            UPDATE users
            SET status = 'premium'
            WHERE created_at < NOW() - INTERVAL 1 YEAR
        """)
    )
    op.execute(
        text("""
            UPDATE users
            SET status = 'new'
            WHERE created_at >= NOW() - INTERVAL 1 YEAR
        """)
    )

def downgrade():
    op.drop_column('users', 'status')

Example 4: Database-Agnostic Migration

Different databases have different SQL syntax. To write a migration that works on PostgreSQL, MySQL, and SQLite, use Alembic's get_context() function to detect the database type:

# migrations/versions/006_database_agnostic_migration.py
from alembic import op, context
import sqlalchemy as sa

def upgrade():
    ctx = context.get_context()
    dialect = ctx.dialect.name  # 'postgresql', 'mysql', 'sqlite'
    
    if dialect == 'postgresql':
        op.execute(sa.text("""
            UPDATE users
            SET first_name = split_part(name, ' ', 1)
        """))
    elif dialect == 'mysql':
        op.execute(sa.text("""
            UPDATE users
            SET first_name = SUBSTRING_INDEX(name, ' ', 1)
        """))
    elif dialect == 'sqlite':
        op.execute(sa.text("""
            UPDATE users
            SET first_name = substr(name, 1, instr(name, ' ') - 1)
        """))

def downgrade():
    op.execute("UPDATE users SET first_name = NULL")

Example 5: Create a Table Manually

If you can't use autogenerate (e.g., for a complex legacy schema), create a table by hand:

# migrations/versions/007_create_posts_table.py
def upgrade():
    op.create_table(
        'posts',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('user_id', sa.Integer(), nullable=False),
        sa.Column('title', sa.String(200), nullable=False),
        sa.Column('body', sa.Text(), nullable=False),
        sa.Column('created_at', sa.DateTime(), default=sa.func.now()),
        sa.PrimaryKeyConstraint('id'),
        sa.ForeignKeyConstraint(['user_id'], ['users.id'], ondelete='CASCADE')
    )
    op.create_index('idx_posts_user_id', 'posts', ['user_id'])

def downgrade():
    op.drop_index('idx_posts_user_id', table_name='posts')
    op.drop_table('posts')

Example 6: Bulk Insert Data

If you need to seed data during a migration (e.g., adding a new status category):

# migrations/versions/008_add_status_categories.py
from sqlalchemy import insert

def upgrade():
    op.create_table(
        'status_categories',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('name', sa.String(50), unique=True)
    )
    
    # Insert seed data
    op.execute(insert(sa.table('status_categories', sa.column('name'))).values([
        ('active',),
        ('inactive',),
        ('suspended',),
    ]))

def downgrade():
    op.drop_table('status_categories')

Testing Your Migration Locally

Before applying to production, test your migration:

# Apply the migration
alembic upgrade head

# Verify the schema changed correctly
psql -U user -d mydb -c "\d users"  # PostgreSQL
# or
sqlite3 app.db ".schema users"  # SQLite

# Rollback to test the downgrade
alembic downgrade -1

# Apply again to confirm
alembic upgrade head

Common Pitfalls

1. Forgetting to make downgrade reversible. If upgrade() adds a column, downgrade() must drop it. Test both directions.
2. Not handling nullable correctly. When adding a column with existing data, make it nullable at first, populate it, then set nullable=False.
3. Assuming database-specific SQL. If your code may run on multiple databases (or you migrate databases later), use dialect checks or keep SQL simple and standard.
4. Not using transactions. Alembic wraps migrations in transactions by default; if a step fails, the whole migration rolls back. Test failures locally.

Key Takeaways

• Use alembic revision -m "message" to create an empty migration file
• Write upgrade() and downgrade() functions using the op object for database-agnostic changes
• Use op.execute(sql) for complex SQL or op.execute(text(...)) for parameterized queries
• Check the database dialect with context.get_context().dialect.name for database-specific logic
• Always test migrations locally: apply, verify, rollback, and re-apply to confirm both directions work
• Keep downgrade reversible so you can confidently roll back in production

Frequently Asked Questions

Can I run Python code in a migration, not just SQL?

Yes. Inside upgrade() and downgrade(), you can import and run any Python code. Get a database connection with op.get_bind() and execute queries manually. However, avoid long-running operations in migrations (they block deployments); keep them focused on schema and essential data changes.

What if my migration fails midway?

Alembic wraps each migration in a database transaction. If any statement fails, the entire migration is rolled back and your schema reverts to the previous state. This is why migrations are safe—they're all-or-nothing.

How do I see the SQL that a migration will run?

Use alembic upgrade head --sql (without actually applying it). This prints the SQL statements that will be executed, useful for reviewing before deploying.

Can I skip a migration if it's already applied?

Use alembic stamp <revision> to mark a specific revision as applied without running it. This is useful if you've manually applied changes to your database and want Alembic to catch up without re-running the migration.

Further Reading

• Alembic Operations Reference
• SQLAlchemy Text and TextClause
• Database-Agnostic SQL Best Practices