Naki-404/naki-postgres

PostgreSQL y SQLAlchemy 2

Cuándo usarlo

• Al diseñar tablas, restricciones e índices.
• Al escribir modelos y consultas de SQLAlchemy 2 (síncronas o asíncronas).
• Cuando una consulta es lenta, se acumulan bloqueos o se agotan las conexiones.
• Al escribir una migración de Alembic, especialmente contra una tabla con tráfico en vivo.

Para cambios destructivos de datos, aplica también naki-safety. Sigue el skill de reglas del proyecto si existe (por ejemplo naki-myapp).

Reglas de esquema

1. Claves primarias: bigint generated always as identity o UUID (v7 si está disponible, para localidad de índices). No expongas identificadores secuenciales públicamente si la enumeración importa.
2. Tipos: timestamptz (nunca timestamp sin zona), text con un CHECK de longitud en lugar de varchar(n) cuando los límites puedan cambiar, numeric para dinero, jsonb (no json), boolean not null default false.
3. Las restricciones son la última línea de defensa: NOT NULL por defecto, UNIQUE, CHECK, claves foráneas con una elección explícita de ON DELETE.
4. Indexa cada clave foránea utilizada en joins o cascadas; Postgres no lo hace automáticamente.
5. Nombra restricciones e índices mediante una convención de nomenclatura en los MetaData de SQLAlchemy para que los diffs de Alembic sean estables.
6. Usa jsonb para datos genuinamente variables, no para evitar modelar. Las claves con muchas consultas se convierten en columnas reales.
7. Eliminación suave solo si la necesitas; si lo haces, agrega índices parciales WHERE deleted_at IS NULL.

Reglas de índices

Patrón de consulta	Índice
igualdad y rango en una columna	B-tree
WHERE a = ? AND b > ? ORDER BY b	compuesto (a, b), columnas de igualdad primero
subconjunto de filas (elementos abiertos)	parcial WHERE status = 'open'
lower(email) = ?	índice de expresión sobre lower(email)
contención jsonb @>	GIN
búsqueda de texto completo, trigrama ILIKE '%x%'	GIN con tsvector o pg_trgm
series temporales enormes de solo inserción	BRIN en la columna de marca de tiempo

• Cada índice cuesta escrituras y espacio. Elimina los no utilizados (pg_stat_user_indexes.idx_scan = 0 durante un período largo).
• Índice de cobertura con INCLUDE (...) para escaneos solo de índice en lecturas frecuentes.

Flujo de trabajo de optimización de consultas

1. Obtén la consulta real (habilita echo o log_min_duration_statement) y los parámetros reales.
2. Ejecuta EXPLAIN (ANALYZE, BUFFERS) en un conjunto de datos similar al de producción. Nunca confíes en planes de una tabla de desarrollo con 100 filas.
3. Busca: Seq Scan en tablas grandes con filtros selectivos, estimación de filas muy alejada de la real (estadísticas obsoletas: ANALYZE), bucles anidados sobre muchas filas, ordenaciones que escriben en disco, 'shared read' alto.
4. Corrige la causa más barata: agrega o ajusta un índice, reescribe el predicado para que sea sargable, elimina N+1, pagina por conjuntos de claves.
5. Vuelve a ejecutar EXPLAIN ANALYZE y compara tiempos. Usa pg_stat_statements para encontrar las consultas con mayor tiempo total.

Reglas de SQLAlchemy 2

1. Usa el estilo 2.0: select(), session.execute(), session.scalars(), Mapped[...] y mapped_column.
2. Una sesión por unidad de trabajo (petición, trabajo). Nunca compartas una sesión entre hilos o tareas.
3. Evita N+1: carga las relaciones explícitamente con selectinload (colecciones) o joinedload (muchos a uno). En código asíncrono, la carga perezosa genera una excepción; establece lazy="raise" en las relaciones para detectarlo también en código síncrono.
4. Usa solo parámetros vinculados. Nunca formatees entrada de usuario en cadenas SQL, incluido text().
5. Configura el tamaño del pool y pool_pre_ping=True; el total de conexiones en todas las instancias de la aplicación debe mantenerse por debajo de max_connections (usa PgBouncer si hay muchos workers).
6. Mantén las transacciones cortas; no hagas llamadas de red a otros servicios mientras se mantiene una transacción.

from datetime import datetime
from sqlalchemy import ForeignKey, Index, String, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, selectinload

class Base(DeclarativeBase):
    pass

class Incident(Base):
    __tablename__ = "incidents"
    id: Mapped[int] = mapped_column(primary_key=True)
    status: Mapped[str] = mapped_column(String(16), index=True)
    opened_at: Mapped[datetime]
    alerts: Mapped[list["Alert"]] = relationship(back_populates="incident", lazy="raise")
    __table_args__ = (Index("ix_incidents_open", "opened_at", postgresql_where="status = 'open'"),)

async def open_incidents(session: AsyncSession, limit: int) -> list[Incident]:
    stmt = (
        select(Incident)
        .where(Incident.status == "open")
        .order_by(Incident.opened_at.desc(), Incident.id.desc())
        .limit(limit)
        .options(selectinload(Incident.alerts))
    )
    return list((await session.scalars(stmt)).all())

Transacciones y bloqueos

• El READ COMMITTED predeterminado es suficiente para la mayoría del trabajo. Usa SELECT ... FOR UPDATE para bloquear filas que modificarás basándote en lo que lees.
• Para colas: FOR UPDATE SKIP LOCKED permite que varios workers reclamen trabajos sin bloquearse entre sí.
• Upserts: INSERT ... ON CONFLICT (key) DO UPDATE en lugar de carreras de leer-then-insertar.
• Establece lock_timeout y statement_timeout para roles de aplicación, para que una consulta defectuosa no pueda bloquear todo.
• Los interbloqueos provienen de bloquear filas en diferentes órdenes; bloquea en un orden consistente (ordena por id).

Alembic y cambios sin tiempo de inactividad

Siempre revisa las migraciones autogeneradas; la autogeneración omite los cambios de nombre (emite drop más add) y algunos cambios de tipo.

Expande y contrae para cambios disruptivos:

1. Expandir: agrega una nueva columna anulable (o con un valor predeterminado constante, lo cual es instantáneo en PG 11+), una nueva tabla o un nuevo índice CONCURRENTLY.
2. Despliega código que escriba tanto lo antiguo como lo nuevo.
3. Rellenar en lotes (por ejemplo, 5k filas por transacción) fuera de la migración, o en una migración de datos separada con registro de progreso.
4. Despliega código que lea lo nuevo.
5. Contraer: elimina la columna antigua en una versión posterior.

Operaciones peligrosas en tablas grandes en vivo:

Operación	Enfoque más seguro
CREATE INDEX	CREATE INDEX CONCURRENTLY (Alembic: postgresql_concurrently=True dentro de with op.get_context().autocommit_block():)
agregar NOT NULL	agregar CHECK (col IS NOT NULL) NOT VALID, VALIDATE CONSTRAINT, luego establecer NOT NULL
agregar clave foránea	agregar con NOT VALID, luego VALIDATE CONSTRAINT
renombrar columna	expandir y contraer, nunca en el lugar con código antiguo ejecutándose
cambiar tipo de columna	nueva columna más relleno, no ALTER TYPE que reescribe la tabla

• Establece SET lock_timeout = '5s' en las migraciones para que fallen rápido en lugar de hacer cola detrás de transacciones largas.
• Cada migración tiene un downgrade() funcional o una nota escrita de por qué no puede tenerlo.
• Prueba las migraciones hacia arriba, hacia abajo y nuevamente hacia arriba en una copia de datos realistas.

Lista de verificación

• timestamptz, NOT NULL por defecto, restricciones nombradas, claves foráneas indexadas.
• Consultas parametrizadas; sin f-strings en SQL.
• Sin N+1; relaciones cargadas explícitamente.
• Consultas lentas verificadas con EXPLAIN (ANALYZE, BUFFERS) en datos realistas.
• Transacciones cortas; tiempos de espera configurados.
• Migración revisada, reversible, usa CONCURRENTLY / NOT VALID donde sea necesario.
• Rellenos por lotes y reanudables.

Errores comunes

• Paginación con OFFSET 100000; usa paginación por conjunto de claves en una clave de ordenación indexada.
• WHERE date(created_at) = ? inhabilitando el índice; usa un rango en la columna original.
• LIKE '%term%' sin índice de trigrama.
• Olvidar await session.commit(), o hacer commit dentro de un bucle por fila.
• Ejecutar alembic revision --autogenerate y confirmarlo sin leerlo.
• Mantener una transacción abierta mientras se llama a una API externa.
• Almacenar marcas de tiempo en hora local.