Explicit SQL escape hatch

SqlDataStorage owns engine lifecycle and opens SQLModel sessions. It does not try to model every aggregate through a generic CRUD abstraction.

When a model requires an exclusion constraint, partial index, window function, or dialect-specific temporal query, dropping to native SQL is an intended use. It is not an abstraction leak when the boundary remains explicit.

The rule

MIC manages persistence plumbing, not the modeling of your aggregates. When the model exceeds the ORM, use native SQL inside the same session and the same 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()

Parameters remain bound and are never interpolated into the SQL string.

Same unit of work

ORM and native SQL can participate in the same 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()

Commit and rollback belong to the session owner. Do not open a second connection inside a repository for advanced SQL: that would leave the domain transaction and break atomicity with the outbox.

Constraints and migrations

Dialect-specific constraints belong in Alembic, not 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 &&
    )
    """
)

The service owns this migration, its target dialect, and its integrity tests. MIC provides the session; it does not simplify a constraint that the database expresses better than the ORM.

Errors and observability

Native SQL does not receive magical error mapping:

• translate only expected constraint violations into DomainError;
• map genuinely transient infrastructure failures to TransientError;
• let programming errors reach operational alerts;
• keep the query under the active span through SQLAlchemy instrumentation or an explicit span, without recording sensitive parameter values.

This is the same policy used for ORM access. Choosing SQL must not hide errors or reduce trace quality.

Explicit limitation

A PostgreSQL-specific query reduces portability to MySQL or SQLite. That can be the correct tradeoff. Make it visible in the repository name, migrations, and integration tests instead of pretending a common abstraction preserves semantics it cannot guarantee.