Migraciones de BD a prueba de fallos: evolución de esquemas con FastAPI × SQLAlchemy × Alembic — autogeneración, ajuste manual, migraciones de datos, transacciones, tests y CI

Resumen (la visión global primero)

• Gestiona versiones de esquema con Alembic y refleja de forma segura cambios en FastAPI + SQLAlchemy.
• Combina autogeneración con ediciones manuales; maneja el cambio entre entornos, scripts de migración de datos y semántica de transacciones.
• De local a producción: listas de verificación, cómo integrarlo en pruebas y CI, y errores operativos comunes.

A quién beneficia

• Aprendiz A (dev en solitario)
  Ajustó modelos y funciona en local, pero falla en otros entornos. Quiere ejecutar migraciones correctamente.
• Equipo pequeño B (taller de 3 personas)
  Cambios de especificación frecuentes; rastrean diffs de BD a mano. Quieren estandarizar autogeneración + revisión.
• Equipo SaaS C (startup)
  Quiere despliegue seguro a staging y producción, y reversión instantánea si algo sale mal.

Evaluación de accesibilidad

• Estructurado con párrafos cortos y viñetas por sección; código monoespaciado con comentarios ligeros.
• Explicaciones breves para términos usados por primera vez; pasos numerados para facilitar el orden.
• Nivel objetivo aproximadamente AA.

1. Supuestos y objetivos

• ORM: estilo SQLAlchemy 2.x; migraciones de esquema con Alembic.
• Puedes aprender con SQLite, pero la paridad de producción asume un RDB como PostgreSQL.
• Objetivo: estandarizar cambio de modelo → generación de diff → revisión → aplicar → verificar para que cualquiera obtenga el mismo resultado.

Puntos de decisión

• Usa siempre una base de datos transaccional en producción.
• La autogeneración es útil, pero los nombres de índices, valores por defecto y funciones del lado del servidor a menudo requieren retoques manuales.

2. Inicializar: integrar Alembic y ponerlo en marcha

2.1 Dependencias

SQLAlchemy>=2.0
alembic
psycopg[binary]  # Ejemplo usando PostgreSQL

2.2 Estructura de directorios (ejemplo)

fastapi-app/
├─ app/
│  ├─ main.py
│  ├─ db.py
│  ├─ models.py
│  └─ settings.py
├─ alembic/
│  ├─ env.py
│  ├─ script.py.mako
│  └─ versions/
└─ alembic.ini

2.3 Comando de init

alembic init alembic

Qué se genera

• alembic.ini incluye una plantilla de URL de BD.
• alembic/env.py es el punto de entrada durante las migraciones. Carga aquí el Base de SQLAlchemy para que la autogeneración “vea” tus tablas objetivo.

3. Configuración de BD en la app y Base

3.1 Settings

# app/settings.py
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
    database_url: str = "postgresql+psycopg://user:pass@localhost:5432/appdb"
    class Config:
        env_file = ".env"
def get_settings() -> Settings:
    return Settings()

3.2 Session y Base

# app/db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

class Base(DeclarativeBase): pass

def get_engine(url: str):
    return create_engine(url, pool_pre_ping=True)

def make_session_factory(url: str):
    return sessionmaker(bind=get_engine(url), autoflush=False, autocommit=False)

3.3 Ejemplo de modelo

# app/models.py
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import String, Integer, DateTime, func
from app.db import Base

class Article(Base):
    __tablename__ = "articles"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    title: Mapped[str] = mapped_column(String(200), index=True, nullable=False)
    body: Mapped[str] = mapped_column(String, nullable=False)
    created_at: Mapped["DateTime"] = mapped_column(DateTime(timezone=True), server_default=func.now(), nullable=False)

4. Conectar env.py con el proyecto FastAPI

Añade código en alembic/env.py para importar el Base de la app y la URL de BD.

# alembic/env.py (extracto)
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context

# 1) Importar el Base y settings de la app
import os, sys
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from app.db import Base
from app.settings import get_settings

config = context.config
if config.get_main_option("sqlalchemy.url") in (None, "", "sqlite:///"):
    config.set_main_option("sqlalchemy.url", get_settings().database_url)

target_metadata = Base.metadata

fileConfig(config.config_file_name)

def run_migrations_offline():
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        compare_type=True,      # detectar cambios de tipo
        compare_server_default=True  # detectar defaults del servidor
    )
    with context.begin_transaction():
        context.run_migrations()

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,
            compare_type=True,
            compare_server_default=True
        )
        with context.begin_transaction():
            context.run_migrations()

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

Puntos clave

• Activa compare_type=True y compare_server_default=True para estabilizar la detección de diffs.
• Obtén la URL desde el entorno o .env, sobreescribiendo los valores por defecto de alembic.ini.

5. Crear la migración inicial

5.1 Crear base

alembic revision --autogenerate -m "init"
# Revisar → aplicar
alembic upgrade head

Se genera alembic/versions/<rev>_init.py. Revisa tablas e índices antes de aplicar.

5.2 Ajustes manuales comunes

• Ten en cuenta las particularidades del SGBD al cambiar longitudes de cadenas (reducir puede perder datos).
• Expresiones como server_default=func.now() difieren por dialecto; especifica a mano si es necesario.
• Para índices compuestos y restricciones únicas, define convenciones de nombres y házlas explícitas.

6. Hacer un cambio y ejecutar diffs

Como ejemplo, añade una columna slug a Article con una restricción única.

6.1 Cambio en el modelo

# app/models.py (adición)
from sqlalchemy import UniqueConstraint
class Article(Base):
    __tablename__ = "articles"
    # columnas existentes omitidas
    slug: Mapped[str] = mapped_column(String(200), nullable=False, unique=True)
    __table_args__ = (UniqueConstraint("slug", name="uq_articles_slug"),)

6.2 Crear diff → revisar → aplicar

alembic revision --autogenerate -m "add slug to articles"
alembic upgrade head

El script contendrá op.add_column y op.create_unique_constraint. Verifica nombres, orden y la sensatez del paso de downgrade.

7. Migraciones de datos (cómo escribirlas)

Si un cambio de esquema necesita valores iniciales o transformaciones para filas existentes, incluye las operaciones de datos en la misma revisión.

# alembic/versions/xxxx_add_slug.py (extracto)
from alembic import op
import sqlalchemy as sa

def upgrade():
    op.add_column("articles", sa.Column("slug", sa.String(length=200), nullable=True))
    # Orden seguro: permitir NULL → rellenar filas existentes → establecer NOT NULL
    conn = op.get_bind()
    articles = conn.execute(sa.text("SELECT id, title FROM articles")).mappings().all()
    for row in articles:
        slug = row["title"].lower().replace(" ", "-")
        conn.execute(sa.text("UPDATE articles SET slug=:s WHERE id=:i"), {"s": slug, "i": row["id"]})
    op.alter_column("articles", "slug", nullable=False)
    op.create_unique_constraint("uq_articles_slug", "articles", ["slug"])

def downgrade():
    op.drop_constraint("uq_articles_slug", "articles", type_="unique")
    op.drop_column("articles", "slug")

Notas

• Con datos existentes, aplica en etapas: añadir columna → backfill → añadir restricciones.
• Actualizaciones grandes pueden causar bloqueos largos; divide en lotes o programa fuera de horas pico.

8. Transacciones y rollback

• En PostgreSQL y similares, cada migración generalmente se ejecuta en una transacción y revierte automáticamente en caso de fallo.
• Para operaciones pesadas (p. ej., cambios de tipo en tablas enormes), investiga el impacto por adelantado.
• El comportamiento de transacciones DDL de SQLite es más débil; los resultados de autogeneración pueden variar con restricciones. Valida en un RDB con paridad de producción.

9. Operar entre entornos (dev/stg/prod)

• dev: crea/aplica migraciones libremente; valida con datos semilla.
• stg: aplica pruebas en BD equivalente a prod; mide duración y tiempo de bloqueo.
• prod: coordina con ventanas de mantenimiento o despliegues escalonados; observa monitores y alertas mientras se ejecuta.

Consejos

• alembic current para la versión actual; alembic history para el historial de revisiones.
• Reversión de emergencia con alembic downgrade -1, pero evita downgrades que pierdan datos. Prefiere correcciones hacia adelante.

10. Manejo de ramas y merges

Cuando varias líneas de desarrollo crean revisiones en paralelo, el grafo de versiones puede bifurcarse.

• Para evitar conflictos, integra a la rama principal con frecuencia y usa alembic revision --autogenerate -m "merge heads" para unificar.
• En la revisión de merge, asegúrate de que ambos diffs se apliquen en el orden correcto.
• Convenciones de nombres estrictas y revisión, y alinear el momento de generación entre devs reduce problemas.

11. ¿Debe FastAPI aplicar migraciones automáticamente al inicio?

• Para desarrollo local, “upgrade head al inicio” es conveniente.
• En producción, prefiere revisión humana + ejecución manual, o un paso explícito en la pipeline de CI/CD.
• Ejecutar alembic upgrade head en el entrypoint del contenedor y abortar el arranque en caso de fallo funciona bien con IaC.

Ejemplo (shell):

set -e
alembic upgrade head
exec uvicorn app.main:app --host 0.0.0.0 --port 8000

12. Integración en pruebas

• Para tests unitarios: crea una BD de test → alembic upgrade head → ejecuta tests → rollback/teardown para dejarla limpia cada vez.
• Para migraciones que tocan datos existentes, prepara datos dummy tipo snapshot y verifica el éxito de upgrade y la consistencia del estado final.
• Confirma reentrancia: fallar y volver a ejecutar no debe corromper resultados.

Borrador con pytest:

import subprocess, os
def setup_module():
    os.environ["DATABASE_URL"] = "postgresql+psycopg://user:pass@localhost:5432/testdb"
    subprocess.check_call(["alembic", "upgrade", "head"])

13. Flujo de CI/CD

• Si un PR incluye alembic revision --autogenerate, haz que CI ejecute alembic upgrade head para validar.
• Tras desplegar a staging, registra tiempo de ejecución y logs de la migración; relaja umbrales de alertas antes de prod.
• Aclara el orden del release: migración de BD → despliegue de la app.

14. Pequeños consejos operativos

• Nombres: que el propósito sea obvio—create_<table>, add_<col>_to_<table>, drop_<col>_from_<table>—para que el historial sea legible.
• Divide cambios grandes en múltiples pasos; no lo hagas todo a la vez.
• Para construcción de índices pesados, considera opciones paralelas/en línea (PostgreSQL CONCURRENTLY), aunque Alembic requerirá SQL manual.
• Mide siempre en una BD de staging cercana al tamaño de producción.

15. Errores comunes y remedios

16. Ejemplo concreto: introducir una tabla de tags y asociaciones

Haremos muchos-a-muchos entre artículos y tags por etapas:

1. Crear tabla tags
2. Crear tabla puente article_tags (PK compuesta)
3. Asociar tags a artículos existentes (migración de datos)
4. Añadir índices necesarios

Cambios de modelo (extracto):

# app/models.py (ejemplo añadido)
from sqlalchemy import Table, Column, ForeignKey
from sqlalchemy.orm import relationship

article_tags = Table(
    "article_tags",
    Base.metadata,
    Column("article_id", ForeignKey("articles.id", ondelete="CASCADE"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id", ondelete="CASCADE"), primary_key=True),
)

class Tag(Base):
    __tablename__ = "tags"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)

class Article(Base):
    __tablename__ = "articles"
    # columnas existentes omitidas
    tags = relationship("Tag", secondary=article_tags, backref="articles")

Genera el diff → revisa índices de la tabla de unión y la política ondelete → añade carga de datos al upgrade si es necesario.

17. Hoja de ruta

1. Inicializa Alembic y ordena env.py.
2. Crea el primer conjunto de tablas vía autogeneración, revisa y aplica.
3. Haz cambios en modelos → genera diffs → ajusta a mano → aplica como estándar del equipo.
4. Escribe migraciones de datos en orden seguro; mide tiempos/bloqueos en tests y staging.
5. Integra con CI/CD; documenta merges de ramas y pasos de respuesta ante emergencias.

Referencias

• Alembic
  • Documentación de Alembic
  • Detalles de Autogenerate
  • Trabajo con ramas
• SQLAlchemy
  • SQLAlchemy 2.0 ORM
  • Migración de 1.x a 2.x
• PostgreSQL
  • Documentación de PostgreSQL
• FastAPI
  • Bases de datos SQL en FastAPI

Cierre

• Versiona tu esquema con Alembic; empieza con autogeneración y termina a mano.
• Con datos existentes, diseña cambios por etapas junto con migraciones de datos; mide siempre el impacto en transacciones/bloqueos.
• Haz que local, staging y producción ejecuten los mismos pasos; integra verificaciones en CI/CD.
• Mantén merges de ramas y convenciones de nombres explícitos para que el historial siga siendo legible. Así, la evolución del modelo se traduce directamente en releases seguros.