Alright, let’s dive into the world of FastAPI and database migrations with Alembic. If you’re building a web app with FastAPI, you’ll want to keep your database schema in sync as your project evolves. That’s where Alembic comes in handy.

First things first, make sure you’ve got FastAPI and SQLAlchemy installed in your project. If not, just run:

pip install fastapi sqlalchemy alembic

Now, let’s set up Alembic in your FastAPI project. Create a new directory called “alembic” in your project root and run:

alembic init alembic

This command creates the necessary files and folders for Alembic to work its magic. You’ll see a new “alembic.ini” file in your project root and an “alembic” folder with some Python files inside.

Next, we need to configure Alembic to work with your database. Open up the “alembic.ini” file and find the line that says “sqlalchemy.url”. Replace it with your database connection string. For example, if you’re using PostgreSQL, it might look like this:

sqlalchemy.url = postgresql://username:password@localhost/mydatabase

Now, let’s create our first migration. In your terminal, run:

alembic revision -m "create users table"

This command creates a new migration file in the “alembic/versions” directory. Open up the newly created file, and you’ll see two functions: “upgrade” and “downgrade”. These functions define how to apply and reverse the migration.

Let’s add some code to create a users table:

from alembic import op
import sqlalchemy as sa

def upgrade():
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('username', sa.String(50), nullable=False),
        sa.Column('email', sa.String(100), nullable=False),
        sa.PrimaryKeyConstraint('id')
    )

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

The “upgrade” function creates the users table, while the “downgrade” function drops it. This allows you to roll back changes if needed.

To apply the migration, run:

alembic upgrade head

This command applies all pending migrations to your database. If you want to roll back the last migration, you can use:

alembic downgrade -1

Now that we’ve got the basics down, let’s integrate Alembic with our FastAPI application. Create a new file called “database.py” in your project root and add the following code:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "postgresql://username:password@localhost/mydatabase"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

This code sets up the database connection and creates a base class for our models.

Next, let’s create a model for our users table. Create a new file called “models.py” and add:

from sqlalchemy import Column, Integer, String
from database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, index=True)
    email = Column(String(100), unique=True, index=True)

Now, whenever we make changes to our models, we can create a new migration to reflect those changes in the database. For example, let’s say we want to add a “created_at” column to our users table. We’d update our User model like this:

from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.sql import func
from database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, index=True)
    email = Column(String(100), unique=True, index=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

Then, we’d create a new migration:

alembic revision -m "add created_at to users"

In the new migration file, we’d add:

from alembic import op
import sqlalchemy as sa

def upgrade():
    op.add_column('users', sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()))

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

And apply the migration with:

alembic upgrade head

Now, let’s talk about some best practices when working with Alembic and FastAPI. First, always make small, incremental changes. It’s easier to manage and debug migrations when they’re focused on a single change.

Second, use meaningful names for your migrations. Instead of “revision_001”, use something like “add_user_profile_table” or “update_product_price_column”. This makes it easier to understand the purpose of each migration at a glance.

Third, test your migrations thoroughly before applying them to production. Create a separate test database and run your migrations there first. This can help catch any potential issues before they affect your live data.

Fourth, consider using Alembic’s branch functionality for feature branches. This allows you to create separate migration paths for different features, which can be especially useful in larger projects with multiple developers.

To create a new branch, use:

alembic revision -m "new feature" --branch-label feature_x

To merge branches, you can use Alembic’s merge command:

alembic merge -m "merge feature_x" feature_x head

Now, let’s talk about how to handle data migrations. Sometimes, you need to not only change the schema but also update existing data. For example, let’s say we want to split the “name” column into “first_name” and “last_name”. We’d create a new migration:

alembic revision -m "split name into first_name and last_name"

And in the migration file:

from alembic import op
import sqlalchemy as sa

def upgrade():
    op.add_column('users', sa.Column('first_name', sa.String(50)))
    op.add_column('users', sa.Column('last_name', sa.String(50)))
    
    # Data migration
    connection = op.get_bind()
    users = connection.execute(sa.text("SELECT id, name FROM users")).fetchall()
    for user in users:
        names = user.name.split(' ', 1)
        first_name = names[0]
        last_name = names[1] if len(names) > 1 else ''
        connection.execute(
            sa.text("UPDATE users SET first_name = :first, last_name = :last WHERE id = :id"),
            first=first_name, last=last_name, id=user.id
        )
    
    op.drop_column('users', 'name')

def downgrade():
    op.add_column('users', sa.Column('name', sa.String(100)))
    
    # Data migration
    connection = op.get_bind()
    users = connection.execute(sa.text("SELECT id, first_name, last_name FROM users")).fetchall()
    for user in users:
        name = f"{user.first_name} {user.last_name}".strip()
        connection.execute(
            sa.text("UPDATE users SET name = :name WHERE id = :id"),
            name=name, id=user.id
        )
    
    op.drop_column('users', 'first_name')
    op.drop_column('users', 'last_name')

This migration not only changes the schema but also updates the existing data to fit the new structure.

One thing to keep in mind when working with Alembic and FastAPI is the concept of “offline” migrations. By default, Alembic generates Python code for migrations, which is then executed against the database. However, in some production environments, you might not have direct access to the database or the ability to run Python code.

In these cases, you can use Alembic’s offline mode to generate SQL scripts instead. To generate an SQL script for all pending migrations, you can use:

alembic upgrade head --sql > migration.sql

This command creates a SQL file with all the necessary database changes, which you can then apply directly to your database using your preferred database management tool.

Another advanced technique is using Alembic’s hooks to perform custom actions before or after migrations. You can define these hooks in your “env.py” file. For example, you might want to disable foreign key checks before a migration and re-enable them afterward:

from alembic import context

def run_migrations_offline():
    # ... existing code ...

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            process_revision_directives=process_revision_directives,
            **current_app.extensions['migrate'].configure_args
        )

        # Disable foreign key checks
        connection.execute("SET FOREIGN_KEY_CHECKS=0;")
        
        with context.begin_transaction():
            context.run_migrations()
        
        # Re-enable foreign key checks
        connection.execute("SET FOREIGN_KEY_CHECKS=1;")

if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

This can be particularly useful when dealing with complex database schemas or when you need to perform specific actions during the migration process.

When working on larger projects, you might want to consider using Alembic’s programmatic API to manage migrations from within your FastAPI application. This can be useful for automating migrations as part of your deployment process or for creating admin interfaces to manage database schema changes.

Here’s an example of how you might use Alembic’s API in a FastAPI endpoint:

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from alembic.config import Config
from alembic import command
from database import SessionLocal

app = FastAPI()

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

@app.post("/migrate")
def run_migrations(db: Session = Depends(get_db)):
    alembic_cfg = Config("alembic.ini")
    command.upgrade(alembic_cfg, "head")
    return {"message": "Migrations applied successfully"}

This endpoint allows you to trigger migrations programmatically, which can be useful in various scenarios.

As your project grows, you might find yourself needing to manage multiple databases or schemas. Alembic supports this use case through its “multiple databases” feature. You can define multiple databases in your “alembic.ini” file and specify which database to target for each migration.

For example, in your “alembic.ini”:

[alembic]
script_location = alembic

[db1]
sqlalchemy.url = postgresql://user:pass@localhost/db1

[db2]
sqlalchemy.url = postgresql://user:pass@localhost/db2

Then, when creating a new migration, you can specify the target database:

alembic revision -m "new migration" --head db1@head

This creates a new migration targeting the “db1” database. You can apply migrations to a specific database using:

alembic upgrade db1@head

This flexibility allows you to manage complex database setups within a single FastAPI project.

In conclusion, integrating Alembic with FastAPI provides a powerful way to manage your database schema changes. By automating your migrations, you can ensure that your database structure stays in sync with your application code, making it easier to develop and maintain your FastAPI projects. Remember to keep your migrations small and focused, test thoroughly before applying to production, and leverage Alembic’s advanced features when needed. Happy coding!