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

greeden

hace 3 meses

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

Síntoma	Causa	Remedio
Autogenerate no ve nada	Base no importada; `target_metadata` sin definir	Asegura que `Base.metadata` se pase en `env.py`
Fallo al reducir longitud de cadena	Datos existentes demasiado largos	Consulta filas que exceden; acuerda política de recorte; migra por etapas
El default del servidor aparece siempre en el diff	Diferencias de dialecto / formas de función	Mantén `compare_server_default`, pero fija nombres de función a mano
Bloqueos largos	Reescritura de tablas enormes	Métodos en línea, ejecuciones por etapas, horas valle, índices concurrentes
Downgrade rompe cosas	Operaciones destructivas	Prefiere correcciones hacia adelante; si el rollback es obligatorio, haz backups antes

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

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

Crear tabla tags
Crear tabla puente article_tags (PK compuesta)
Asociar tags a artículos existentes (migración de datos)
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

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

Referencias

Alembic
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.

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)

A quién beneficia

Evaluación de accesibilidad

1. Supuestos y objetivos

2. Inicializar: integrar Alembic y ponerlo en marcha

2.1 Dependencias

2.2 Estructura de directorios (ejemplo)

2.3 Comando de init

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

3.1 Settings

3.2 Session y Base

3.3 Ejemplo de modelo

4. Conectar env.py con el proyecto FastAPI

5. Crear la migración inicial

5.1 Crear base

5.2 Ajustes manuales comunes

6. Hacer un cambio y ejecutar diffs

6.1 Cambio en el modelo

6.2 Crear diff → revisar → aplicar

7. Migraciones de datos (cómo escribirlas)

8. Transacciones y rollback

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

10. Manejo de ramas y merges

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

12. Integración en pruebas

13. Flujo de CI/CD

14. Pequeños consejos operativos

15. Errores comunes y remedios

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

17. Hoja de ruta

Referencias

Cierre

Comparte esto:

4. Conectar `env.py` con el proyecto FastAPI