Naki-404/naki-postgres

PostgreSQL et SQLAlchemy 2

Quand l'utiliser

• Conception de tables, contraintes et index.
• Écriture de modèles et requêtes SQLAlchemy 2, synchrones ou asynchrones.
• Une requête est lente, les verrous s'accumulent ou les connexions sont épuisées.
• Écriture d'une migration Alembic, en particulier sur une table avec du trafic en direct.

Pour les modifications destructrices de données, appliquez également naki-safety. Suivez la compétence des règles du projet si le projet en possède une (par exemple naki-myapp).

Règles de schéma

1. Clés primaires : bigint generated always as identity ou UUID (v7 si disponible, pour la localité des index). N'exposez pas d'identifiants séquentiels publiquement si le dénombrement est important.
2. Types : timestamptz (jamais timestamp sans fuseau horaire), text avec un CHECK sur la longueur au lieu de varchar(n) lorsque les limites peuvent changer, numeric pour l'argent, jsonb (pas json), boolean not null default false.
3. Les contraintes sont la dernière ligne de défense : NOT NULL par défaut, UNIQUE, CHECK, clés étrangères avec un choix explicite de ON DELETE.
4. Indexez chaque clé étrangère utilisée dans les jonctions ou les cascades ; Postgres ne le fait pas automatiquement.
5. Nommez les contraintes et les index via une convention de nommage dans MetaData de SQLAlchemy pour que les diffs Alembic soient stables.
6. jsonb pour les données vraiment variables, pas pour éviter la modélisation. Les clés fréquemment interrogées deviennent de vraies colonnes.
7. Suppression logicielle uniquement si nécessaire ; si c'est le cas, ajoutez des index partiels WHERE deleted_at IS NULL.

Règles des index

Modèle de requête	Index
Égalité et intervalle sur une colonne	B-tree
WHERE a = ? AND b > ? ORDER BY b	composite (a, b), colonnes d'égalité en premier
Sous-ensemble de lignes (éléments ouverts)	partiel WHERE status = 'open'
lower(email) = ?	index d'expression sur lower(email)
Contenu jsonb @>	GIN
Recherche plein texte, trigramme ILIKE '%x%'	GIN avec tsvector ou pg_trgm
Énormes séries temporelles append-only	BRIN sur l'horodatage

• Chaque index coûte en écriture et en disque. Supprimez ceux qui sont inutilisés (pg_stat_user_indexes.idx_scan = 0 sur une longue période).
• Index de couverture avec INCLUDE (...) pour les analyses index-only sur les lectures à chaud.

Flux de travail de réglage des requêtes

1. Obtenez la requête réelle (activez echo ou log_min_duration_statement) et les paramètres réels.
2. Exécutez EXPLAIN (ANALYZE, BUFFERS) sur un jeu de données proche de la production. Ne faites jamais confiance aux plans d'une table de développement avec 100 lignes.
3. Recherchez : Seq Scan sur les grandes tables avec des filtres sélectifs, estimation du nombre de lignes loin de la réalité (statistiques obsolètes : ANALYZE), boucles imbriquées sur de nombreuses lignes, tris débordant sur le disque, shared read élevé.
4. Corrigez la cause la moins coûteuse : ajoutez ou ajustez un index, réécrivez le prédicat pour le rendre sargable, supprimez le N+1, paginez par keyset.
5. Relancez EXPLAIN ANALYZE et comparez les temps. Utilisez pg_stat_statements pour trouver les requêtes les plus coûteuses en temps total.

Règles SQLAlchemy 2

1. Utilisez le style 2.0 : select(), session.execute(), session.scalars(), Mapped[...] et mapped_column.
2. Une session par unité de travail (requête, tâche). Ne partagez jamais une session entre des threads ou des tâches.
3. Évitez le N+1 : chargez les relations explicitement avec selectinload (collections) ou joinedload (plusieurs-vers-un). En code asynchrone, le chargement paresseux lève une exception ; définissez lazy="raise" sur les relations pour le détecter également en code synchrone.
4. Utilisez uniquement des paramètres liés. Ne formatez jamais une entrée utilisateur dans des chaînes SQL, y compris text().
5. Définissez la taille du pool et pool_pre_ping=True ; le nombre total de connexions sur toutes les instances de l'application doit rester inférieur à max_connections (utilisez PgBouncer si vous avez de nombreux workers).
6. Gardez les transactions courtes ; pas d'appels réseau vers d'autres services pendant qu'une transaction est ouverte.

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

Transactions et verrouillage

• Le niveau READ COMMITTED par défaut convient à la plupart des travaux. Utilisez SELECT ... FOR UPDATE pour verrouiller les lignes que vous allez modifier en fonction de ce que vous avez lu.
• Pour les files d'attente : FOR UPDATE SKIP LOCKED permet à plusieurs workers de réclamer des tâches sans se bloquer mutuellement.
• Upserts : INSERT ... ON CONFLICT (key) DO UPDATE au lieu de courses lecture-puis-insertion.
• Définissez lock_timeout et statement_timeout pour les rôles applicatifs afin qu'une mauvaise requête ne puisse pas tout bloquer.
• Les interblocages surviennent lorsque les lignes sont verrouillées dans un ordre différent ; verrouillez dans un ordre cohérent (triez par id).

Alembic et modifications sans temps d'arrêt

Passez toujours en revue les migrations autogénérées ; l'autogénération manque les renommages (elle émet une suppression plus un ajout) et certains changements de type.

Étendez et contractez pour les modifications cassantes :

1. Étendez : ajoutez une nouvelle colonne nullable (ou avec une valeur par défaut constante, ce qui est instantané sur PG 11+), une nouvelle table ou un nouvel index CONCURRENTLY.
2. Déployez le code qui écrit à la fois l'ancien et le nouveau.
3. Rattrapage par lots (par exemple 5 000 lignes par transaction) en dehors de la migration, ou dans une migration de données séparée avec journalisation de la progression.
4. Déployez le code qui lit le nouveau.
5. Contractez : supprimez l'ancienne colonne dans une version ultérieure.

Opérations dangereuses sur les grandes tables en direct :

Opération	Approche plus sûre
CREATE INDEX	CREATE INDEX CONCURRENTLY (Alembic : postgresql_concurrently=True dans with op.get_context().autocommit_block():)
ajouter NOT NULL	ajouter CHECK (col IS NOT NULL) NOT VALID, VALIDATE CONSTRAINT, puis définir NOT NULL
ajouter une clé étrangère	ajouter avec NOT VALID, puis VALIDATE CONSTRAINT
renommer une colonne	étendre et contracter, jamais en place avec l'ancien code en cours d'exécution
changer le type d'une colonne	nouvelle colonne plus rattrapage, pas ALTER TYPE qui réécrit la table

• Définissez SET lock_timeout = '5s' dans les migrations pour qu'elles échouent rapidement au lieu de faire la queue derrière de longues transactions.
• Chaque migration a une downgrade() fonctionnelle ou une note écrite expliquant pourquoi elle n'est pas possible.
• Testez les migrations en montée, descente et remontée sur une copie de données réalistes.

Liste de vérification

• timestamptz, valeurs par défaut NOT NULL, contraintes nommées, clés étrangères indexées.
• Requêtes paramétrées ; pas de f-strings dans le SQL.
• Pas de N+1 ; relations chargées explicitement.
• Requêtes lentes vérifiées avec EXPLAIN (ANALYZE, BUFFERS) sur des données réalistes.
• Transactions courtes ; délais d'attente définis.
• Migration revue, réversible, utilise CONCURRENTLY / NOT VALID si nécessaire.
• Rattrapages par lots et reprenables.

Erreurs courantes

• Pagination avec OFFSET 100000 ; utilisez keyset sur une clé de tri indexée.
• WHERE date(created_at) = ? qui contourne l'index ; utilisez un intervalle sur la colonne brute.
• LIKE '%term%' sans index trigramme.
• Oublier await session.commit(), ou valider dans une boucle par ligne.
• Exécuter alembic revision --autogenerate et le valider sans l'avoir lu.
• Maintenir une transaction ouverte tout en appelant une API externe.
• Stocker les horodatages en heure locale.