Mini Project: CI/CD Migration Pipeline

DodaTech Updated 2026-06-28 11 min read

In this tutorial, you will learn about Mini Project: CI/CD Migration Pipeline. We cover key concepts, practical examples, and best practices to help you master this topic.

Build a complete CI/CD migration pipeline: implement automated database migrations with Alembic, GitHub Actions, staging verification, rollback automation, monitoring, and team notifications for safe production schema changes.

What You Learn

You will build a complete CI/CD migration pipeline from scratch: set up Alembic with auto-generation, write migration tests in CI, implement a deployment pipeline with migrate/deploy/verify stages, add automated rollback on failure, configure team notifications, and practice end-to-end schema evolution.

Why It Matters

A production-ready migration pipeline prevents schema-related incidents. Git-based workflows with CI/CD validation ensure migration correctness before deployment. Automated rollback reduces downtime when things go wrong. This project gives you hands-on experience building the pipeline that keeps production databases safe.

Real-World Use

DodaTech's migration pipeline processes 15+ migrations per week across 5 environments. The pipeline has prevented 23 production incidents in 18 months. Every migration is tested in CI, deployed with rollback capability, and monitored for issues. This project teaches you to build the same pipeline.

Project Overview

graph LR
    subgraph "Development"
        Code[Write Code] --> Migrate[Create Migration]
        Migrate --> Test[Test Locally]
    end
    subgraph "CI Pipeline"
        Test --> CI[Push to GitHub]
        CI --> FreshDB[Create Fresh DB]
        FreshDB --> Apply[Apply Migrations]
        Apply --> Verify[Verify Schema]
        Verify --> Rollback[Test Rollback]
    end
    subgraph "CD Pipeline"
        Rollback --> Deploy[Merge to Main]
        Deploy --> Staging[Deploy to Staging]
        Staging --> Production[Deploy to Production]
        Production --> Monitor[Monitor]
    end

The project has three phases: development (create and test migrations locally), CI (automated testing in GitHub Actions), and CD (automated deployment with rollback).

Step 1: Initialize the Project

# Create project directory
mkdir migration-pipeline-project
cd migration-pipeline-project

# Initialize Python project
python3 -m venv venv
source venv/bin/activate

# Install dependencies
pip install alembic sqlalchemy psycopg2-binary pytest

# Initialize Alembic
alembic init alembic

# Output:
# Creating directory alembic/versions... done
# Creating alembic/env.py... done
# Creating alembic/script.py.mako... done
# Creating alembic.ini... done

# app/models.py
"""Application models."""
from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql import func

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    email = Column(String(255), unique=True, nullable=False)
    name = Column(String(200))
    created_at = Column(DateTime, server_default=func.now())

class Order(Base):
    __tablename__ = 'orders'

    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
    total = Column(Float, nullable=False)
    status = Column(String(20), server_default='pending')
    created_at = Column(DateTime, server_default=func.now())

Expected output: Project is initialized with Alembic, SQLAlchemy models, and a virtual environment. Users and Orders models define the initial schema.

Step 2: Configure Alembic

# alembic/env.py
"""Alembic environment configuration."""
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context
import os

from app.models import Base

config = context.config

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

config.set_main_option('sqlalchemy.url', os.environ.get(
    'DATABASE_URL',
    'postgresql://postgres:postgres@localhost/migration_pipeline'
))

target_metadata = Base.metadata

def run_migrations_offline():
    """Run migrations in offline mode."""
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online():
    """Run migrations in online mode."""
    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,
        )
        with context.begin_transaction():
            context.run_migrations()

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

Expected output: Alembic env.py is configured to use the application's SQLAlchemy models for auto-generation. The DATABASE_URL environment variable specifies the target database.

Step 3: Create the Initial Migration

# Generate initial migration from models
alembic revision --autogenerate -m "initial_schema"

# Output:
# Generating /alembic/versions/abc123_initial_schema.py... done

# Apply the migration
alembic upgrade head

# Output:
# Running upgrade -> abc123, initial_schema

# alembic/versions/abc123_initial_schema.py
# Auto-generated migration
"""initial_schema"""
from alembic import op
import sqlalchemy as sa

revision = 'abc123'
down_revision = None

def upgrade():
    op.create_table('users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('email', sa.String(length=255), nullable=False),
        sa.Column('name', sa.String(length=200), nullable=True),
        sa.Column('created_at', sa.DateTime(), server_default=sa.func.now()),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('email'),
    )
    op.create_table('orders',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('user_id', sa.Integer(), nullable=False),
        sa.Column('total', sa.Float(), nullable=False),
        sa.Column('status', sa.String(length=20), server_default='pending'),
        sa.Column('created_at', sa.DateTime(), server_default=sa.func.now()),
        sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
        sa.PrimaryKeyConstraint('id'),
    )

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

Expected output: Initial migration creates users and orders tables. Auto-generation detected both models and created the migration. Both upgrade and downgrade functions are generated.

Step 4: Write Migration Tests

# tests/test_migrations.py
"""Test suite for database migrations."""
import os
import pytest
from sqlalchemy import create_engine, text
from alembic.config import Config
from alembic.command import upgrade, downgrade, current

TEST_DB_URL = os.environ.get(
    'TEST_DATABASE_URL',
    'postgresql://postgres:postgres@localhost/test_pipeline'
)

@pytest.fixture
def alembic_config():
    """Create Alembic configuration pointing to test database."""
    config = Config('alembic.ini')
    config.set_main_option('sqlalchemy.url', TEST_DB_URL)
    return config

@pytest.fixture(autouse=True)
def clean_database():
    """Ensure a clean database before each test."""
    engine = create_engine(TEST_DB_URL)
    with engine.begin() as conn:
        conn.execute(text("DROP TABLE IF EXISTS alembic_version CASCADE"))
        conn.execute(text("DROP TABLE IF EXISTS orders CASCADE"))
        conn.execute(text("DROP TABLE IF EXISTS users CASCADE"))
    engine.dispose()

class TestMigrations:
    def test_apply_all_migrations(self, alembic_config):
        """Verify all migrations apply cleanly from scratch."""
        upgrade(alembic_config, 'head')
        current_revision = current(alembic_config)
        assert current_revision is not None, "Migrations should be applied"

    def test_rollback_and_reapply(self, alembic_config):
        """Verify rollback and re-apply works."""
        upgrade(alembic_config, 'head')
        downgrade(alembic_config, '-1')
        upgrade(alembic_config, 'head')
        current_revision = current(alembic_config)
        assert current_revision is not None, "Migrations should re-apply"

    def test_schema_has_expected_tables(self, alembic_config):
        """Verify expected tables exist after migration."""
        upgrade(alembic_config, 'head')
        engine = create_engine(TEST_DB_URL)
        with engine.connect() as conn:
            tables = conn.execute(
                text("SELECT table_name FROM information_schema.tables "
                     "WHERE table_schema = 'public'")
            ).fetchall()
            table_names = [t[0] for t in tables]
            assert 'users' in table_names, "users table should exist"
            assert 'orders' in table_names, "orders table should exist"
        engine.dispose()

    def test_schema_has_expected_columns(self, alembic_config):
        """Verify users table has expected columns."""
        upgrade(alembic_config, 'head')
        engine = create_engine(TEST_DB_URL)
        with engine.connect() as conn:
            columns = conn.execute(
                text("SELECT column_name FROM information_schema.columns "
                     "WHERE table_name = 'users'")
            ).fetchall()
            column_names = [c[0] for c in columns]
            assert 'email' in column_names
            assert 'name' in column_names
            assert 'created_at' in column_names
        engine.dispose()

Expected output: Migration tests verify clean application, rollback capability, and expected schema structure. Tests run against a clean test database every time.

Step 5: Set Up CI Pipeline

# .github/workflows/ci.yml
name: CI - Migration Tests

on:
    pull_request:
        paths:
            - 'alembic/**'
            - 'app/models.py'
            - 'requirements.txt'

jobs:
    migration-tests:
        runs-on: ubuntu-latest

        services:
            postgres:
                image: postgres:16
                env:
                    POSTGRES_PASSWORD: postgres
                ports:
                    - 5432:5432
                options: >-
                    --health-cmd pg_isready
                    --health-interval 10s
                    --health-timeout 5s
                    --health-retries 5

        steps:
            - uses: actions/checkout@v4

            - name: Set up Python
              uses: actions/setup-python@v5
              with:
                  python-version: '3.12'

            - name: Install dependencies
              run: |
                  python -m pip install --upgrade pip
                  pip install -r requirements.txt

            - name: Create test database
              run: createdb -h localhost -U postgres test_pipeline

            - name: Run migration tests
              run: pytest tests/ -v
              env:
                  TEST_DATABASE_URL: postgresql://postgres:postgres@localhost/test_pipeline
                  DATABASE_URL: postgresql://postgres:postgres@localhost/test_pipeline

Expected output: CI pipeline runs migration tests on every PR that changes migrations or models. Services section starts PostgreSQL. Tests run against a fresh database.

Step 6: Set Up CD Pipeline

# .github/workflows/cd.yml
name: CD - Migration Deployment

on:
    push:
        branches:
            - main

jobs:
    test:
        runs-on: ubuntu-latest
        services:
            postgres:
                image: postgres:16
                env:
                    POSTGRES_PASSWORD: postgres
                ports:
                    - 5432:5432
                options: >-
                    --health-cmd pg_isready
                    --health-interval 10s
                    --health-timeout 5s
                    --health-retries 5

        steps:
            - uses: actions/checkout@v4
            - name: Set up Python
              uses: actions/setup-python@v5
              with:
                  python-version: '3.12'
            - name: Install dependencies
              run: pip install -r requirements.txt
            - name: Create test database
              run: createdb -h localhost -U postgres test_pipeline
            - name: Run migration tests
              run: pytest tests/ -v
              env:
                  TEST_DATABASE_URL: postgresql://postgres:postgres@localhost/test_pipeline
                  DATABASE_URL: postgresql://postgres:postgres@localhost/test_pipeline

    migrate-staging:
        needs: [test]
        runs-on: ubuntu-latest
        environment: staging
        steps:
            - uses: actions/checkout@v4
            - name: Set up Python
              uses: actions/setup-python@v5
              with:
                  python-version: '3.12'
            - name: Install dependencies
              run: pip install -r requirements.txt
            - name: Run migrations on staging
              run: alembic upgrade head
              env:
                  DATABASE_URL: ${{ secrets.STAGING_DATABASE_URL }}

    migrate-production:
        needs: [migrate-staging]
        runs-on: ubuntu-latest
        environment: production
        steps:
            - uses: actions/checkout@v4
            - name: Set up Python
              uses: actions/setup-python@v5
              with:
                  python-version: '3.12'
            - name: Install dependencies
              run: pip install -r requirements.txt
            - name: Backup database
              run: pg_dump $DATABASE_URL > backup_$(date +%Y%m%d_%H%M%S).sql
              env:
                  DATABASE_URL: ${{ secrets.PRODUCTION_DATABASE_URL }}
            - name: Run migrations on production
              run: alembic upgrade head
              env:
                  DATABASE_URL: ${{ secrets.PRODUCTION_DATABASE_URL }}
            - name: Verify migration
              run: python -c "from alembic.config import Config; from alembic.command import current; c = Config('alembic.ini'); c.set_main_option('sqlalchemy.url', '${{ secrets.PRODUCTION_DATABASE_URL }}'); current(c)"

Expected output: CD pipeline runs tests first, then deploys to staging, then to production. Production migration includes a backup step. Each environment is a separate job with appropriate secrets.

Step 7: Add Rollback and Monitoring

# .github/workflows/rollback.yml
name: Rollback Migration

on:
    workflow_dispatch:
        inputs:
            steps:
                description: 'Number of migrations to rollback'
                required: true
                default: '1'
            environment:
                description: 'Target environment'
                required: true
                default: 'production'

jobs:
    rollback:
        runs-on: ubuntu-latest
        environment: ${{ github.event.inputs.environment }}
        steps:
            - uses: actions/checkout@v4
            - name: Set up Python
              uses: actions/setup-python@v5
              with:
                  python-version: '3.12'
            - name: Install dependencies
              run: pip install -r requirements.txt
            - name: Take backup before rollback
              run: pg_dump $DATABASE_URL > pre_rollback_backup.sql
              env:
                  DATABASE_URL: ${{ secrets[format('{0}_DATABASE_URL', github.event.inputs.environment)] }}
            - name: Rollback migrations
              run: alembic downgrade -${{ github.event.inputs.steps }}
              env:
                  DATABASE_URL: ${{ secrets[format('{0}_DATABASE_URL', github.event.inputs.environment)] }}
            - name: Verify rollback
              run: alembic current
              env:
                  DATABASE_URL: ${{ secrets[format('{0}_DATABASE_URL', github.event.inputs.environment)] }}
            - name: Notify team
              run: |
                  curl -X POST -H 'Content-type: application/json' \
                  --data '{"text":"Rollback of ${{ github.event.inputs.steps }} migration(s) on ${{ github.event.inputs.environment }} completed"}' \
                  ${{ secrets.SLACK_WEBHOOK_URL }}

Expected output: Rollback workflow is triggered manually with parameters (steps and environment). It takes a backup, runs the rollback, verifies the current version, and notifies the team.

Step 8: Make a Schema Change

# Step 1: Add a new column to the User model
# app/models.py (updated)
class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    email = Column(String(255), unique=True, nullable=False)
    name = Column(String(200))
    phone = Column(String(20), nullable=True)  # New column
    created_at = Column(DateTime, server_default=func.now())

# Step 2: Auto-generate the migration
alembic revision --autogenerate -m "add_phone_to_users"

# Output:
# Generating /alembic/versions/def456_add_phone_to_users.py... done

# Step 3: Review the generated migration
# alembic/versions/def456_add_phone_to_users.py
"""add_phone_to_users"""
revision = 'def456'
down_revision = 'abc123'

def upgrade():
    op.add_column('users', sa.Column('phone', sa.String(20), nullable=True))

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

# Step 4: Test locally
alembic upgrade head
alembic downgrade -1
alembic upgrade head

# Step 5: Push to GitHub
git add .
git commit -m "Add phone column to users"
git push origin feature/add-phone

Expected output: Adding a column follows the complete workflow: modify model, auto-generate migration, review, test locally, commit, and push. CI tests run automatically.

Step 9: Test the Full Pipeline

#!/bin/bash
# test_pipeline.sh - End-to-end pipeline test
set -euo pipefail

echo "=== Migration Pipeline End-to-End Test ==="

# Step 1: Create fresh test database
echo "Creating test database..."
createdb pipeline_e2e_test
export DATABASE_URL=postgresql://postgres:postgres@localhost/pipeline_e2e_test

# Step 2: Run tests
echo "Running migration tests..."
pytest tests/ -v

# Step 3: Apply all migrations
echo "Applying all migrations..."
alembic upgrade head

# Step 4: Verify schema
echo "Verifying schema..."
python -c "
from sqlalchemy import create_engine, text
engine = create_engine('$DATABASE_URL')
with engine.connect() as conn:
    tables = conn.execute(text(\"SELECT table_name FROM information_schema.tables WHERE table_schema='public'\")).fetchall()
    print(f'Tables: {[t[0] for t in tables]}')
"

# Step 5: Test rollback
echo "Testing rollback..."
alembic downgrade -1
echo "Rolled back one migration"
alembic upgrade head
echo "Re-applied migration"

# Step 6: Insert test data
echo "Testing data operations..."
python -c "
from sqlalchemy import create_engine, text
engine = create_engine('$DATABASE_URL')
with engine.begin() as conn:
    conn.execute(text(\"INSERT INTO users (email, name) VALUES ('test@test.com', 'Test User')\"))
    result = conn.execute(text('SELECT * FROM users'))
    print(f'User created: {result.fetchone()}')
"

echo "=== Pipeline Test Passed ==="

# Cleanup
dropdb pipeline_e2e_test

Expected output: End-to-end test verifies the complete pipeline: tests pass, migrations apply, schema is correct, rollback works, and data operations succeed.

Practice Questions

1. What are the stages of the migration pipeline in this project?

Development (modify models, auto-generate migration, test locally), CI (GitHub Actions tests migrations against fresh database), CD (deploy to staging, then production with backup), and Rollback (manual workflow with backup and notification).

2. Why does the CD pipeline run migrations on staging before production?

Staging deployment catches environment-specific issues before production. If a migration fails on staging, it is fixed before reaching production. Staging mirrors production configuration closely.

3. What is the purpose of the rollback workflow?

The rollback workflow provides a safe, automated way to revert migrations when issues are discovered. It takes a backup before rolling back, verifies the rollback, and notifies the team.

4. How does the CI pipeline ensure migration correctness?

CI creates a fresh database, applies all migrations, runs tests that verify schema structure and rollback capability, and runs the application test suite. Every PR with migration changes runs this validation.

Challenge

Extend the pipeline project with: a data migration (backfill a new column from existing data), a squashing migration (combine the two migrations into one baseline), a zero-downtime deployment Strategy (expand-contract pattern), a feature flag integration (toggle between old and new schema), and a monitoring dashboard that shows migration status across environments.

FAQ

How do I adapt this pipeline for MySQL instead of PostgreSQL?

Change the PostgreSQL service to MySQL in CI, update the DATABASE_URL format (mysql://user:pass@host/db), and ensure the MySQL driver (mysqlclient or pymysql) is installed. Alembic supports MySQL natively.

Can I use this pipeline with Prisma instead of Alembic?

Yes. Replace Alembic commands with Prisma commands: npx prisma migrate dev, npx prisma migrate deploy, npx prisma migrate status. The pipeline structure remains the same.

How do I handle secrets securely in the pipeline?

Use GitHub Secrets for database credentials. Store DATABASE_URL, backup credentials, and webhook URLs as secrets. Never commit secrets to the repository. Use environment-specific secrets for staging and production.

What if my migration takes longer than the CI timeout?

Increase the CI job timeout. For migrations over 30 minutes, consider running them outside CI (dedicated migration runner). Use the expand-contract pattern to minimize long migrations.

How do I add database backup to the pipeline?

Add a backup step before migration execution. Use pg_dump for PostgreSQL, mysqldump for MySQL. Store backups in cloud storage (S3, GCS) with retention policies. Test backup restoration periodically.

Can I run this pipeline for multiple databases?

Use a matrix strategy in GitHub Actions. Define the databases in a matrix (postgres, mysql, sqlite). Run migration tests against each database type. Deployment targets the production database type.

What's Next

Now that you have built a complete CI/CD migration pipeline, explore other backend topics like Message Queue Patterns for asynchronous processing, or Celery for task queue management.

← Previous Golang Migrate for Go — Complete Guide Next → Liquibase for Java — Complete Guide

Built by the developers of DodaTech

Doda Browser, DodaZIP & Durga Antivirus Pro

Home Browse Database Migrations