Automated Database Migrations in CI/CD Pipelines

Database migrations are one of the trickiest parts of automated deployments. Run them too early and your app crashes against an old schema; run them too late and you've already rolled out incompatible application code. Here's how to get it right.

The Expand-Contract Pattern

The safest migration strategy in zero-downtime deployments: (1) Expand—add new columns/tables as nullable, keeping the old structure intact. (2) Deploy new application code that writes to both old and new columns. (3) Migrate—backfill data. (4) Contract—remove old columns once all app instances are updated.

Alembic in CI/CD (Flask)

# In GitHub Actions deploy step
- name: Run Database Migrations
  run: |
    flask db upgrade
    echo "Migrations applied successfully"

Always run flask db upgrade before starting new application containers. Use a dedicated migration job in your Docker Compose or Kubernetes manifests that completes before the app pods start.

Kubernetes Init Containers for Migrations

initContainers:
  - name: run-migrations
    image: ghcr.io/mirahlabs/api:latest
    command: ["flask", "db", "upgrade"]
    envFrom:
      - secretRef:
          name: app-secrets

Rollback Strategy

Maintain a downgrade() function in every Alembic migration. Before any production deploy, verify the downgrade script works in staging. For destructive operations (DROP COLUMN), add a 2-sprint safety window before the final contract step.

Production Database Connection Pool Manager

Here is an optimized database transaction manager in Python utilizing SQLAlchemy 2.0 with explicit connection pooling limits, connection recycling, and automated deadlock retries:

import logging
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy.exc import DBAPIError

logger = logging.getLogger("MirahLabs.DatabasePool")
DATABASE_URL = "postgresql://user:pass@localhost:5432/db"

engine = create_engine(
    DATABASE_URL,
    pool_size=20,
    max_overflow=10,
    pool_recycle=1800,
    pool_timeout=30,
    pool_pre_ping=True
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def run_transaction_with_retry(session_action_func, max_retries=3):
    for attempt in range(max_retries):
        db = SessionLocal()
        try:
            result = session_action_func(db)
            db.commit()
            return result
        except DBAPIError as e:
            db.rollback()
            if attempt == max_retries - 1: raise e
            logger.warning(f"Deadlock detected. Retrying attempt {attempt+2}...")
        finally:
            db.close()

Cloud Infrastructure Performance Profile

Below is a comparative latency and throughput profile of this infrastructure pattern deployed under a simulated load of 10,000 concurrent requests:

US & UK Compliance and Data Governance

Modern applications operating across US and UK regions must establish comprehensive data governance frameworks. This includes meeting the security baselines of the US NIST Cybersecurity Framework and the UK Cyber Essentials certification. Enforcing encryption at rest and in transit, keeping audit logs, and maintaining a clear incident response plan are essential to comply with both CCPA and UK GDPR regulations.