Sortie SQL explicite

SqlDataStorage gère le cycle de vie de l'engine et ouvre des sessions SQLModel. Il ne cherche pas à modéliser tous les agrégats avec un CRUD générique.

Quand un modèle exige une contrainte d'exclusion, un index partiel, une fonction de fenêtre ou une requête temporelle spécifique au dialecte, descendre au SQL natif est une utilisation prévue. Ce n'est pas une fuite d'abstraction tant que la frontière reste explicite.

La règle

MIC gère la plomberie de persistance, pas la modélisation de vos agrégats. Quand le modèle dépasse l'ORM, utilisez le SQL natif dans la même session et la même transaction.

from sqlalchemy import text
from mic.datastorage import SqlDataStorage


storage = SqlDataStorage(url=settings.database_url)

with storage.session() as session:
    rows = session.execute(
        text(
            """
            SELECT resource_id, field_id, value
            FROM resource_facts
            WHERE resource_id = :resource_id
              AND valid_from <= :valid_at
              AND (valid_to IS NULL OR valid_to > :valid_at)
              AND known_from <= :known_at
              AND (known_to IS NULL OR known_to > :known_at)
            """
        ),
        {
            "resource_id": resource_id,
            "valid_at": valid_at,
            "known_at": known_at,
        },
    ).mappings().all()

Les paramètres restent liés, jamais interpolés dans la chaîne SQL.

Même unité de travail

ORM et SQL natif peuvent participer à la même transaction :

with storage.session() as session:
    session.add(resource_record)
    session.execute(
        text("UPDATE counters SET value = value + 1 WHERE name = :name"),
        {"name": "resources"},
    )
    session.commit()

Le commit et le rollback appartiennent au propriétaire de la session. N'ouvrez pas une seconde connexion dans un repository pour exécuter le SQL avancé : elle sortirait de la transaction métier et casserait l'atomicité avec l'outbox.

Contraintes et migrations

Les contraintes dialectales appartiennent à Alembic, pas à create_all() :

from alembic import op


op.execute(
    """
    ALTER TABLE resource_facts
    ADD CONSTRAINT resource_facts_no_overlap
    EXCLUDE USING gist (
      resource_id WITH =,
      field_id WITH =,
      tstzrange(valid_from, valid_to) WITH &&,
      tstzrange(known_from, known_to) WITH &&
    )
    """
)

Le service possède cette migration, son dialecte cible et ses tests d'intégrité. MIC fournit la session ; il ne simplifie pas une contrainte que la base sait exprimer mieux que l'ORM.

Erreurs et observabilité

Le SQL natif ne bénéficie pas d'un mapping magique :

• traduisez uniquement les violations attendues en DomainError ;
• mappez les indisponibilités réellement transitoires en TransientError ;
• laissez les erreurs de programmation remonter vers les alertes ;
• conservez la requête dans le span actif avec une instrumentation SQLAlchemy ou un span explicite, sans enregistrer les valeurs de paramètres sensibles.

Cette politique est la même que pour l'ORM. Le choix de SQL ne doit ni masquer une erreur ni réduire la qualité des traces.

Limite assumée

Une requête PostgreSQL spécifique réduit la portabilité vers MySQL ou SQLite. C'est parfois le bon compromis. Rendez-le visible dans le nom du repository, les migrations et les tests d'intégration au lieu de prétendre qu'une abstraction commune conserve des sémantiques qu'elle ne peut pas garantir.