Database migration in Python using SQLAlchemy and Alembic

Pre-requisite

1. Package Manager uv. If not already installed, install
2. Clone the repository. For the rest of the documentation, it will be assumed the code is cloned in MIGRATION_CODE folder.
3. Run uv sync to create virtual environment with all required libraries

$ uv sync

4. Activate the environment

$ source .venv/bin/activate

NOTE

This is tested only in Ubuntu.

How Alembic works for Database migration

In order to understand that, let us go step by step starting from scratch.

Step : Basic Set up

First, we created an alembic structure.

$ cd MIGRATTION_CODE/migration
$ alembic run alembic

This will create a scaffolding for alembic to work.

alembic
| -- versions
| -- env.py
| -- README
| -- script.py.mako
alembic.ini

[here an __init__.py is also added manually within alembic folder]

Next step is to modify alembic.ini and alembic->env.py file.

In alembic.ini, we add the line [Line 43]

version_locations = alembic/versions

This instructs alembic to use the versions folder to store the migration scripts (we will see this later).

We also change the URL of the database we want to interact with using alembic

sqlalchemy.url = sqlite:///../sample_database.db

This will point to a sqlite database located in $MIGRATION_CODE. We can replace it with other drivers as needed. For example, in general

sqlalchemy.url = driver://user:pass@localhost/dbname

However, we may not want to use pass in plain text. Another alternative is to remove the sqlalchemy_url from alembic.ini and rather use alembic->env.py to specify the uri. After the config object is created. we can add the uri as below:

config.set_main_option('sqlalchemy.url', <db_uri>)

where <db_uri> can be loaded from environment or other sources (e.g. secret manager etc.)

This completes the basic set up of alembic

Step : Create empty schema

[This is available in commit 64b67437b60bf1ec38c9ad8e82bb57bf2f318d75]

Let us create a new module db_models and put in an __init__.py. We will also create a base.py with following entry

from sqlalchemy.orm import declarative_base

Base = declarative_base()

Then we import this Base object in alembic->env.py

from migration.db_models import Base

and add this to the same file

target_metadata = Base.metadata

At this point, if we execute

$ alembic revision --autogenerate -m "first empty schema"

a new migration script will be auto-generated under alembic->versions.

"""First empty schema created

Revision ID: e74ac50f5bd5
Revises: 
Create Date: 2025-10-16 22:07:43.900102

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = 'e74ac50f5bd5'
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    pass
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    pass
    # ### end Alembic commands ###

This script does not do much because we have not defined any tables within the schema.

However, few things to note:

• revision points to a revision of the database
• down_revision is None because there was no prior revision
• upgrade() and downgrade() are empty.

If we now execute

$ alembic upgrade head

a new empty database will be created in MIGRATION_CODE/sample_database.db

Step: Create first table

[Refer to commit e4342d4a245973d98374f43bb47e3093cc21451a] We add a User Table in db_models. Create a user.pyand define auserstable usingsqlalchemy` model.

Let us run again the autogenerate command

$ alembic revision --autogenerate -m "user table created"

It will create another migration script.

# revision identifiers, used by Alembic.
revision: str = 'cf3883b80c44'
down_revision: Union[str, None] = 'e74ac50f5bd5'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('users',
    sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
    sa.Column('first_name', sa.String(length=50), nullable=False),
    sa.Column('last_name', sa.String(length=50), nullable=False),
    sa.Column('age', sa.Integer(), nullable=True),
    sa.PrimaryKeyConstraint('id')
    )
    op.create_index(op.f('ix_users_id'), 'users', ['id'], unique=False)
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_index(op.f('ix_users_id'), table_name='users')
    op.drop_table('users')
    # ### end Alembic commands ###

Now we see something interesting.

• revision points to this revision.
• down_revision points the revision just before this one (in our case, the empty revision)
• upgrade() method is populated with Operations that correspond to the insertion of the new table and index.
• downgrade() method is poulated with Operations to invert the changes in this revision.

Again running alembic upgrade head will apply these changes and a new table will be added to the database.

Step Modifying the table

[see commit 806d4f862e45d5162f5e61daac73cea00b0c1e37] Now let us assume we changed the table definition and wanted to replace two columns (first_name and last_name) to a single column full_name. We make the necessary changes to the user.py. Now running

$ alembic revision --autogenerate -m "user table modified"

creates a new migration script.

# revision identifiers, used by Alembic.
revision: str = 'b65721ca074e'
down_revision: Union[str, None] = 'cf3883b80c44'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('users', sa.Column('full_name', sa.String(length=100), nullable=False))
    op.drop_column('users', 'first_name')
    op.drop_column('users', 'last_name')
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('users', sa.Column('last_name', sa.VARCHAR(length=50), nullable=False))
    op.add_column('users', sa.Column('first_name', sa.VARCHAR(length=50), nullable=False))
    op.drop_column('users', 'full_name')
    # ### end Alembic commands ###

We can again see the down_revision points to the earlier version. upgrade() and downgrade() methods are appropriately updated. Running alemic upgrade head will update the table accordingly.

Making changes manually

The steps above show how to use Alembic to auto-detect changes in ORM and create migration scripts automatically. But there will be use when we want to force manual changes (e.g adding fact rows etc.). In these cases, we can create an empty migration script by executing

$ alembic revision -m "Some manual changes".

A new migration script created as following:

# revision identifiers, used by Alembic.
revision: str = '135c95539d5b'
down_revision: Union[str, None] = 'b65721ca074e'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    pass


def downgrade() -> None:
    pass

We can see it has already populated revision and down_revision parameters. Now we need to simply fill up the upgrade() and downgrade() methods.