De SQLite a PostgreSQL: 39 modelos ORM y 1.204 filas

SQLite es una base de datos notable. Alimenta mas aplicaciones que cualquier otro motor de base de datos en el mundo. Para prototipado, es imbatible: cero configuracion, un solo archivo y soporte SQL completo. Comenzamos 0fee.dev con SQLite porque nos permitio construir a maxima velocidad con cero sobrecarga de infraestructura.

Pero SQLite tiene una limitacion fundamental que lo hace inadecuado para una plataforma de pagos en produccion: solo permite un escritor a la vez. Cuando solicitudes API concurrentes intentan insertar transacciones, actualizar estados y registrar entregas de webhooks simultaneamente, se serializan. Bajo carga, este cuello de botella de escritor unico se convierte en un muro.

En la sesion 081, migramos 0fee.dev de SQLite a PostgreSQL. 39 modelos ORM de SQLAlchemy. 1.204 filas de datos a traves de 39 tablas. Cada consulta SQL sin procesar convertida a ORM. Y un conjunto de errores que solo salieron a la superficie gracias a la migracion.

Por que la migracion era necesaria

Las condiciones de carrera WAL documentadas en el articulo 059 fueron el detonante inmediato. Pero las razones arquitectonicas eran mas profundas:

La limitacion de escritura concurrente fue el factor decisivo. Una plataforma de pagos procesa pagos, registra webhooks, actualiza estados y genera facturas simultaneamente. Serializar esas escrituras era una bomba de relojeria.

Los 39 modelos ORM

La migracion requirio convertir cada tabla a un modelo ORM de SQLAlchemy apropiado. Muchas rutas habian estado usando SQL sin procesar -- un patron que funcionaba con SQLite pero necesitaba cambiar independientemente del backend de base de datos (para prevencion de inyeccion SQL entre otras razones).

Usando agentes Claude en paralelo, convertimos las 39 tablas en una sola sesion:

python# Ejemplo: modelo Transaction (uno de 39)
from sqlalchemy import (
    Column, String, Float, DateTime, ForeignKey,
    JSON, Boolean, Integer, func
)
from sqlalchemy.orm import relationship
from models.base import Base

class Transaction(Base):
    __tablename__ = "transactions"

    id = Column(String, primary_key=True)
    app_id = Column(String, ForeignKey("apps.id", ondelete="RESTRICT"), nullable=False)
    user_id = Column(String, ForeignKey("users.id"), nullable=False)

    # Campos de monto (post-actualizacion de moneda)
    source_amount = Column(Float, nullable=False)
    source_currency = Column(String(3), nullable=False)
    destination_amount = Column(Float, nullable=True)
    destination_currency = Column(String(3), nullable=True)

    # Detalles del pago
    provider = Column(String, nullable=True)
    payment_method = Column(String, nullable=True)
    status = Column(String, default="pending")
    reference = Column(String, nullable=False)

    # Datos del proveedor
    provider_transaction_id = Column(String, nullable=True)
    meta = Column(JSON, nullable=True)  # Renombrado de 'metadata'

    # Marcas de tiempo
    created_at = Column(DateTime, server_default=func.now())
    updated_at = Column(DateTime, onupdate=func.now())
    completed_at = Column(DateTime, nullable=True)

    # Relaciones
    app = relationship("App", back_populates="transactions")
    user = relationship("User", back_populates="transactions")
    events = relationship("PaymentEvent", back_populates="transaction")
    webhook_deliveries = relationship("WebhookDelivery", back_populates="transaction")

El inventario completo de modelos:

Convirtiendo SQL sin procesar a ORM

La parte mas tediosa de la migracion fue convertir las consultas SQL sin procesar a consultas ORM de SQLAlchemy. Aqui hay un ejemplo representativo:

python# ANTES: SQL sin procesar
async def get_user_transactions(user_id: str, status: str = None):
    query = f"""
        SELECT t.*, a.name as app_name
        FROM transactions t
        JOIN apps a ON t.app_id = a.id
        WHERE t.user_id = ?
    """
    params = [user_id]
    if status:
        query += " AND t.status = ?"
        params.append(status)
    query += " ORDER BY t.created_at DESC"

    result = await db.execute(text(query), params)
    return result.fetchall()

python# DESPUES: SQLAlchemy ORM
async def get_user_transactions(user_id: str, status: str = None):
    query = (
        select(Transaction)
        .options(joinedload(Transaction.app))
        .where(Transaction.user_id == user_id)
    )
    if status:
        query = query.where(Transaction.status == status)
    query = query.order_by(Transaction.created_at.desc())

    result = await db.scalars(query)
    return result.all()

La version ORM es segura en tipos, inmune a inyeccion SQL y produce el mismo SQL. Pero la conversion no siempre fue directa -- consultas complejas con multiples JOINs, GROUP BYs y subconsultas requirieron traduccion cuidadosa.

Migracion de datos: 1.204 filas

La migracion de los datos en si fue manejada por un script Python que leia de SQLite y escribia en PostgreSQL:

python# scripts/migrate_sqlite_to_postgres.py
import sqlite3
import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

SQLITE_PATH = "data/0fee.db"
POSTGRES_URL = "postgresql+asyncpg://user:pass@localhost:5432/zerofee"

# Orden de migracion de tablas (respeta claves foraneas)
MIGRATION_ORDER = [
    "users",
    "apps",
    "providers",
    "payment_methods",
    "app_providers",
    "api_keys",
    "transactions",
    "payment_events",
    "webhooks",
    "webhook_deliveries",
    # ... 29 tablas restantes
]

async def migrate_table(table_name: str, sqlite_conn, pg_session: AsyncSession):
    """Migrar una sola tabla de SQLite a PostgreSQL."""
    cursor = sqlite_conn.execute(f"SELECT * FROM {table_name}")
    columns = [desc[0] for desc in cursor.description]
    rows = cursor.fetchall()

    if not rows:
        print(f"  {table_name}: 0 rows (empty)")
        return 0

    # Manejar renombrado de columna: metadata -> meta
    if "metadata" in columns:
        idx = columns.index("metadata")
        columns[idx] = "meta"

    # Construir sentencia INSERT
    model = get_model_for_table(table_name)
    for row in rows:
        data = dict(zip(columns, row))
        obj = model(**data)
        pg_session.add(obj)

    await pg_session.flush()
    print(f"  {table_name}: {len(rows)} rows migrated")
    return len(rows)

async def run_migration(): sqlite_conn = sqlite3.connect(SQLITE_PATH) pg_engine = create_async_engine(POSTGRES_URL) BLANK async with AsyncSession(pg_engine) as session: total = 0 for table in MIGRATION_ORDER: count = await migrate_table(table, sqlite_conn, session) total += count BLANK await session.commit() print(f"\nTotal: {total} rows migrated across {len(MIGRATION_ORDER)} tables") BLANK sqlite_conn.close() ```

El orden de migracion era critico. Las restricciones de clave foranea en PostgreSQL son estrictas -- no puedes insertar una transaccion que referencia una app inexistente. El script procesa las tablas en orden de dependencia: usuarios primero, luego apps, luego transacciones.

Estadisticas finales de migracion:

users:              23 rows
apps:               31 rows
providers:          53 rows
payment_methods:   117 rows
app_providers:      47 rows
api_keys:           38 rows
transactions:      312 rows
payment_events:    487 rows
webhooks:           19 rows
webhook_deliveries: 34 rows
... (29 more tables)

Total: 1,204 rows across 39 tables

Errores revelados por la migracion

Persistencia de branding

La configuracion de branding (colores, logos, texto personalizado para paginas de checkout) se almacenaba de una manera que asumia los tipos flexibles de SQLite. Al moverla a los tipos estrictos de PostgreSQL, varios campos fallaron:

python# SQLite aceptaba esto (almacena como cadena)
branding.primary_color = 0xFF6B35  # Entero almacenado como cadena

# PostgreSQL lo rechazo
# Column 'primary_color' expects VARCHAR, got INTEGER

La correccion fue asegurar que todos los valores de branding fueran explicitamente convertidos a cadenas antes del almacenamiento.

Tipos Float de PayinMethods

La configuracion de metodos de pago incluia montos minimos y maximos almacenados como enteros en SQLite. La aplicacion de tipos mas estricta de PostgreSQL revelo que algunos valores deberian haber sido flotantes:

python# ANTES: Montos enteros causaban truncamiento
class PayinMethod(Base):
    min_amount = Column(Integer)  # 0.50 USD almacenado como 0
    max_amount = Column(Integer)  # 999.99 almacenado como 999

python# DESPUES: Montos flotantes preservan la precision
class PayinMethod(Base):
    min_amount = Column(Float, default=0.0)
    max_amount = Column(Float, nullable=True)  # None = sin limite

Este era un error sutil que habia estado corrompiendo datos silenciosamente en SQLite. Los montos minimos de $0.50 se almacenaban como $0, desactivando efectivamente la validacion de monto minimo.

El renombrado de metadata a meta

Como se cubrio en el articulo de SQLAdmin, el nombre de columna metadata entra en conflicto con el atributo interno MetaData de SQLAlchemy. El script de migracion manejo este renombrado, pero requirio actualizar cada consulta, cada respuesta de API, cada SDK y cada pieza de documentacion que referenciaba el campo.

Pool de conexiones con psycopg2

Las conexiones a PostgreSQL son costosas de crear. A diferencia de SQLite (que abre un archivo), cada conexion a PostgreSQL implica handshake TCP, autenticacion y negociacion de protocolo. El pool de conexiones reutiliza conexiones establecidas:

python# database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,          # Mantener 10 conexiones inactivas
    max_overflow=20,       # Permitir hasta 20 adicionales bajo carga
    pool_timeout=30,       # Esperar 30s por una conexion antes de error
    pool_recycle=3600,     # Reciclar conexiones despues de 1 hora
    pool_pre_ping=True,    # Verificar salud de conexion antes de usar
)

async_session = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

async def get_db() -> AsyncSession:
    async with async_session() as session:
        try:
            yield session
        finally:
            await session.close()

La configuracion pool_pre_ping=True es importante para produccion. Envia una consulta ligera (SELECT 1) antes de devolver una conexion del pool, asegurando que las conexiones obsoletas (eliminadas por timeouts de red, reinicios del servidor) sean detectadas y reemplazadas.

Driver asincrono: asyncpg

Usamos asyncpg como el driver asincrono de PostgreSQL, que es significativamente mas rapido que psycopg2 para cargas de trabajo asincronas:

python# Formato de URL de conexion para asyncpg
DATABASE_URL = "postgresql+asyncpg://zerofee:password@localhost:5432/zerofee"

# vs psycopg2 (sincrono)
# DATABASE_URL = "postgresql+psycopg2://zerofee:password@localhost:5432/zerofee"

El driver asyncpg es una implementacion pura de Python que usa el protocolo binario de PostgreSQL, haciendolo mas rapido y completamente compatible con asyncio.

Probando la migracion

Validamos la migracion con tres verificaciones:

python# 1. Verificacion de conteo de filas
async def verify_row_counts():
    sqlite_counts = get_sqlite_counts()
    pg_counts = await get_postgres_counts()

    for table in MIGRATION_ORDER:
        assert sqlite_counts[table] == pg_counts[table], \
            f"Row count mismatch in {table}: SQLite={sqlite_counts[table]}, PG={pg_counts[table]}"

# 2. Verificacion de datos de muestra (verificacion puntual de 10 filas por tabla)
async def verify_sample_data():
    for table in MIGRATION_ORDER:
        sqlite_rows = get_sqlite_sample(table, 10)
        pg_rows = await get_pg_sample(table, 10)
        for s_row, p_row in zip(sqlite_rows, pg_rows):
            assert normalize(s_row) == normalize(p_row), \
                f"Data mismatch in {table}"

# 3. Pruebas de humo de endpoints API
async def smoke_test_endpoints():
    endpoints = [
        "/api/health",
        "/api/providers",
        "/api/payment-methods",
    ]
    for endpoint in endpoints:
        response = await client.get(endpoint)
        assert response.status_code == 200, f"Failed: {endpoint}"

Lo que aprendimos

Comienza con PostgreSQL si estas construyendo una plataforma de pagos. SQLite nos ahorro tiempo en las primeras sesiones pero nos costo una sesion completa para la migracion mas horas depurando problemas de WAL. Los tipos estrictos de PostgreSQL habrian detectado errores antes. Su soporte de escritura concurrente habria eliminado toda una clase de condiciones de carrera.

Los modelos ORM valen la inversion inicial. Convertir 39 tablas de SQL sin procesar a ORM fue tedioso pero transformador. Elimino riesgos de inyeccion SQL, hizo el codigo seguro en tipos y habilito la integracion con SQLAdmin. Si hubieramos comenzado con ORM, la migracion habria sido un cambio de cadena de conexion.

Los tipos estrictos son una funcionalidad, no una limitacion. Los tipos flexibles de SQLite ocultaban errores. Las columnas enteras aceptaban silenciosamente flotantes, las columnas de cadena aceptaban silenciosamente enteros. La negativa de PostgreSQL a aceptar discrepancias de tipo nos obligo a corregir problemas de integridad de datos que no sabiamos que teniamos.

Prueba la migracion con conteos de filas, verificaciones puntuales y pruebas de humo. Confia, pero verifica. Una migracion que se completa sin errores no es necesariamente correcta. Verifica que los conteos de filas coincidan, que los datos de muestra coincidan y que los endpoints de API devuelvan resultados correctos.

Este articulo es parte de la serie "Como construimos 0fee.dev". 0fee.dev es un orquestador de pagos que cubre mas de 53 proveedores en mas de 200 paises, construido por Juste A. GNIMAVO y Claude desde Abiyan sin ningun ingeniero humano. Sigue la serie para conocer la historia completa de construccion.