SQLAlchemy 2.0 async

L'ORM typé de SQLAlchemy 2.0

Le style 2.0 est déclaratif ET typé — mypy voit les colonnes :

class Variant(Base):
    __tablename__ = "variant"
    id:        Mapped[uuid.UUID] = mapped_column(primary_key=True)
    tenant_id: Mapped[uuid.UUID]
    name:      Mapped[str]
    price:     Mapped[float | None]          # nullable visible dans le type !

Les requêtes sont des expressions select() exécutées sur une AsyncSession (driver asyncpg) :

async with session_factory() as session:
    rows = (await session.execute(
        select(Variant).where(Variant.name.ilike(f"%{q}%")).limit(20)
    )).scalars().all()

Les trois réglages du pool qui comptent

• pool_pre_ping=True : teste la connexion avant usage — élimine les « server closed the connection » après une nuit calme ;
• expire_on_commit=False : les objets restent lisibles après commit (sinon chaque accès post-commit refait un SELECT surprise — piège n°1 en async) ;
• le search_path est passé au driver au démarrage : le schéma (copilot) n'est jamais écrit en dur dans les requêtes — il vient de la config (DJUST_DB_SCHEMA).

Et le piège async classique : le lazy loading implicite des relations ne fonctionne pas en async (pas d'I/O implicite possible) — les relations se chargent explicitement (selectinload), ce qui a le mérite de rendre les N+1 visibles au lieu de silencieux.

Les migrations : du SQL nu, discipliné

Pas d'Alembic : un runner maison applique des fichiers SQL horodatés (202605060903_variant.sql), chacun idempotent (CREATE TABLE IF NOT EXISTS, DROP POLICY IF EXISTS avant CREATE POLICY) avec SET LOCAL lock_timeout pour ne jamais bloquer la prod, et un linter SQL (squawk) en CI. La philosophie : les features PostgreSQL utilisées (RLS, colonnes générées, HNSW…) sont trop spécifiques pour une couche d'abstraction — le SQL est assumé comme code de premier rang.