Naki-404/naki-postgres

PostgreSQL e SQLAlchemy 2

Quando usar

• Projetando tabelas, restrições e índices.
• Escrevendo modelos e consultas SQLAlchemy 2, síncronas ou assíncronas.
• Uma consulta está lenta, bloqueios se acumulam ou as conexões se esgotam.
• Escrevendo uma migração Alembic, especialmente contra uma tabela com tráfego ao vivo.

Para alterações destrutivas de dados, também aplique naki-safety. Siga a skill de regras do projeto se o projeto tiver uma (por exemplo naki-myapp).

Regras de esquema

1. Chaves primárias: bigint generated always as identity ou UUID (v7 se disponível, para localidade do índice). Não exponha ids sequenciais publicamente se a enumeração for importante.
2. Tipos: timestamptz (nunca timestamp sem fuso horário), text com um CHECK no comprimento em vez de varchar(n) quando os limites podem mudar, numeric para dinheiro, jsonb (não json), boolean not null default false.
3. Restrições são a última linha de defesa: NOT NULL por padrão, UNIQUE, CHECK, chaves estrangeiras com uma escolha explícita de ON DELETE.
4. Indexe toda chave estrangeira usada em junções ou cascatas; o Postgres não faz isso automaticamente.
5. Nomeie restrições e índices através de uma convenção de nomenclatura no MetaData do SQLAlchemy para que os diffs do Alembic sejam estáveis.
6. jsonb para dados genuinamente variáveis, não para evitar a modelagem. Chaves consultadas com frequência tornam-se colunas reais.
7. Soft delete apenas se precisar; se precisar, adicione índices parciais WHERE deleted_at IS NULL.

Regras de índices

Padrão de consulta	Índice
igualdade e intervalo em uma coluna	B-tree
WHERE a = ? AND b > ? ORDER BY b	composto (a, b), colunas de igualdade primeiro
subconjunto de linhas (itens abertos)	parcial WHERE status = 'open'
lower(email) = ?	índice de expressão em lower(email)
contenção jsonb @>	GIN
texto completo, trigrama ILIKE '%x%'	GIN com tsvector ou pg_trgm
série temporal enorme somente para anexação	BRIN no timestamp

• Cada índice custa escritas e disco. Remova os não utilizados (pg_stat_user_indexes.idx_scan = 0 ao longo de uma longa janela).
• Índice de cobertura com INCLUDE (...) para varreduras somente de índice em leituras intensas.

Fluxo de otimização de consultas

1. Obtenha a consulta real (ative echo ou log_min_duration_statement) e os parâmetros reais.
2. Execute EXPLAIN (ANALYZE, BUFFERS) em um conjunto de dados semelhante ao de produção. Nunca confie em planos de uma tabela de desenvolvimento com 100 linhas.
3. Procure por: Seq Scan em tabelas grandes com filtros seletivos, estimativa de linhas muito distante da real (estatísticas desatualizadas: ANALYZE), loops aninhados sobre muitas linhas, ordenações derramando para disco, shared read alto.
4. Corrija a causa mais barata: adicione ou ajuste o índice, reescreva o predicado para ser sargável, remova N+1, pagine por keyset.
5. Reexecute EXPLAIN ANALYZE e compare os tempos. Use pg_stat_statements para encontrar as consultas com maior tempo total.

Regras do SQLAlchemy 2

1. Use o estilo 2.0: select(), session.execute(), session.scalars(), Mapped[...] e mapped_column.
2. Uma sessão por unidade de trabalho (requisição, job). Nunca compartilhe uma sessão entre threads ou tarefas.
3. Evite N+1: carregue relacionamentos explicitamente com selectinload (coleções) ou joinedload (muitos-para-um). Em código assíncrono, o carregamento preguiçoso gera erro; defina lazy="raise" nos relacionamentos para capturá-lo também em código síncrono.
4. Use apenas parâmetros vinculados. Nunca formate entrada do usuário em strings SQL, incluindo text().
5. Defina o tamanho do pool e pool_pre_ping=True; o total de conexões em todas as instâncias do aplicativo deve permanecer abaixo de max_connections (use PgBouncer se houver muitos workers).
6. Mantenha as transações curtas; nenhuma chamada de rede para outros serviços enquanto mantém uma transação.

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())

Transações e bloqueios

• O padrão READ COMMITTED é adequado para a maioria dos trabalhos. Use SELECT ... FOR UPDATE para bloquear linhas que você modificará com base no que leu.
• Para filas: FOR UPDATE SKIP LOCKED permite que vários workers reivindiquem jobs sem se bloquearem mutuamente.
• Upserts: INSERT ... ON CONFLICT (key) DO UPDATE em vez de corridas de leitura-seguida-de-inserção.
• Defina lock_timeout e statement_timeout para funções de aplicativo, para que uma consulta ruim não possa parar tudo.
• Deadlocks vêm do bloqueio de linhas em ordens diferentes; bloqueie em uma ordem consistente (classifique por id).

Alembic e alterações com zero downtime

Sempre revise as migrações geradas automaticamente; a autogeração não detecta renomeações (ela emite drop mais add) e algumas alterações de tipo.

Expanda e contraia para alterações de quebra:

1. Expandir: adicione nova coluna anulável (ou com um padrão constante, que é instantâneo no PG 11+) , nova tabela ou novo índice CONCURRENTLY.
2. Implante código que escreva tanto no antigo quanto no novo.
3. Backfill em lotes (por exemplo, 5k linhas por transação) fora da migração, ou em uma migração de dados separada com registro de progresso.
4. Implante código que leia o novo.
5. Contrair: remova a coluna antiga em uma versão posterior.

Operações perigosas em tabelas grandes e ativas:

Operação	Abordagem mais segura
CREATE INDEX	CREATE INDEX CONCURRENTLY (Alembic: postgresql_concurrently=True dentro de with op.get_context().autocommit_block():)
adicionar NOT NULL	adicione CHECK (col IS NOT NULL) NOT VALID, VALIDATE CONSTRAINT, então defina NOT NULL
adicionar chave estrangeira	adicione com NOT VALID, depois VALIDATE CONSTRAINT
renomear coluna	expanda e contraia, nunca no local com código antigo em execução
alterar tipo de coluna	nova coluna mais backfill, não ALTER TYPE reescrevendo a tabela

• Defina SET lock_timeout = '5s' nas migrações para que falhem rapidamente em vez de enfileirar atrás de transações longas.
• Cada migração tem um downgrade() funcional ou uma nota escrita explicando por que não pode.
• Teste as migrações para cima, para baixo e para cima novamente em uma cópia de dados realistas.

Checklist

• timestamptz, padrões NOT NULL, restrições nomeadas, chaves estrangeiras indexadas.
• Consultas parametrizadas; sem f-strings em SQL.
• Sem N+1; relacionamentos carregados explicitamente.
• Consultas lentas verificadas com EXPLAIN (ANALYZE, BUFFERS) em dados realistas.
• Transações curtas; timeouts definidos.
• Migração revisada, reversível, usa CONCURRENTLY / NOT VALID onde necessário.
• Backfills em lote e retomáveis.

Erros comuns

• Paginação com OFFSET 100000; use keyset em uma chave de ordenação indexada.
• WHERE date(created_at) = ? anulando o índice; use um intervalo na coluna bruta.
• LIKE '%term%' sem índice trigrama.
• Esquecer await session.commit(), ou fazer commit dentro de um loop por linha.
• Executar alembic revision --autogenerate e commitá-lo sem ler.
• Manter uma transação aberta enquanto chama uma API externa.
• Armazenar timestamps em hora local.