Rewrite merge methods on real Core constructs so schema_translate_map applies to loaded data

.github/workflows/ci.yml

@@ -6,18 +6,20 @@ on:
 jobs:
   label-gate:
     uses: AustralianCancerDataNetwork/cava-devops/.github/workflows/label-gate.yml@main
-  build-test:
-    uses: AustralianCancerDataNetwork/cava-devops/.github/workflows/build-test-postgres.yml@main
+  build-test-sqlite:
+    uses: AustralianCancerDataNetwork/cava-devops/.github/workflows/build-test.yml@main
+  build-test-postgres:
+    uses: AustralianCancerDataNetwork/cava-devops/.github/workflows/build-test-postgres-v2.yml@main
     with:
       postgres-db: orm_loader_test
       setup-commands: |
         uv run omop-config configure orm_loader \
-          --set test_orm_db.kind=cdm \
-          --set test_orm_db.connection.dialect=postgresql+psycopg \
-          --set test_orm_db.connection.host=localhost \
-          --set test_orm_db.connection.port=5432 \
-          --set test_orm_db.connection.user=test \
-          --set test_orm_db.connection.password=test \
-          --set test_orm_db.connection.database_name=orm_loader_test \
-          --set test_orm_db.connection.test_only=true \
-          --set test_orm_db.schema_name=public
+          --set test_orm_db_pg.kind=cdm \
+          --set test_orm_db_pg.connection.dialect=postgresql+psycopg \
+          --set test_orm_db_pg.connection.host=localhost \
+          --set test_orm_db_pg.connection.port=5432 \
+          --set test_orm_db_pg.connection.user=test \
+          --set test_orm_db_pg.connection.password=test \
+          --set test_orm_db_pg.connection.database_name=orm_loader_test \
+          --set test_orm_db_pg.connection.test_only=true \
+          --set test_orm_db_pg.cdm_schema=public

docs/tables/mat_view.md

@@ -105,18 +105,38 @@ PatientSummaryMV.refresh_mv(engine, concurrently=True)
 
 This is fail-closed by design. An index created manually outside `__mv_indexes__` does not satisfy the mixin's declaration contract; declare it in the class even if another migration is responsible for creating it. Expressions, partial indexes, and other index forms are outside this simple contract and should not be represented as `MaterializedViewIndex` entries.
 
-By default, `schema=None` leaves the view name unqualified. PostgreSQL resolves that name through the connection's `search_path`, matching the behavior of existing callers. Pass `schema="reporting"` only when the caller intentionally wants an explicit schema-qualified target.
+The view's physical schema comes from the bound connection's own `schema_translate_map`. `create_mv()`/`refresh_mv()`/`drop_mv()` resolve it via `oa_configurator.schema_of()` before ever reaching the backend, which only ever sees an already-resolved physical schema string.
+
+The schema tag itself comes from one of three places, in order:
+
+1. an explicit `schema_tag=` argument, for a one-off override;
+2. when the class is also declaratively mapped (combined with a `Base`, with a real `__table__`), that table's own `schema` -- set the schema the same way as any other mapped table, via `__table_args__ = {"schema": ...}`, and the materialized view follows it automatically;
+3. `__mv_schema_tag__` (defaults to `"primary"`), only consulted for a Core-only declaration with no mapped table to derive anything from.
 
 ```python
-# Existing/default behavior: search_path resolves the target.
+# Core-only: no Base, no mapped table, so __mv_schema_tag__ is what resolves it.
+class VocabSummaryMV(MaterializedViewMixin):
+    __mv_name__ = "vocab_summary"
+    __mv_select__ = ...
+    __mv_schema_tag__ = Role.VOCAB.value  # resolves via the connection's vocab schema
+
+# Uses __mv_schema_tag__ ("primary" by default) via the connection's own schema_translate_map.
 RecentObservationMV.create_mv(engine)
 
-# Explicit schema: the target is quoted and schema-qualified.
-RecentObservationMV.create_mv(engine, schema="reporting")
-RecentObservationMV.refresh_mv(engine, schema="reporting")
+# Override for one call.
+RecentObservationMV.create_mv(engine, schema_tag=Role.VOCAB.value)
+
+# Declaratively mapped: the mapped table's own schema is authoritative, no
+# __mv_schema_tag__ needed (or consulted) at all.
+class VocabPatientSummaryMV(Base, MaterializedViewMixin):
+    __mv_name__ = "vocab_patient_summary"
+    __mv_select__ = ...
+    __tablename__ = "vocab_patient_summary"
+    __table_args__ = {"schema": Role.VOCAB.value}
+    patient_id = sa.Column(sa.Integer, primary_key=True)
 ```
 
-Explicit schema targets are quoted component by component. This matters for embedded quotes, spaces, and mixed-case identifiers. It also means an unqualified mixed-case name and the same name passed with `schema=` can address different PostgreSQL relations. Keep schema selection at the call site and do not assume that this API provides `schema_translate_map`, role-token, or general multi-schema behavior.
+Every generated identifier is quoted through `oa_configurator.qualified()`, which quotes each component only when the dialect actually requires it (reserved words, mixed case, embedded quotes or spaces) — the same behavior every other Core-built query in this stack has.
 
 ## Failure handling and backend support
 
@@ -166,7 +186,3 @@ The built-in implementation is PostgreSQL-oriented. SQLite rejects materialized-
 ::: orm_loader.mappers.ConcurrentRefreshNotEligibleError
     options:
       heading_level: 3
-
-::: orm_loader.mappers.UnsupportedMaterializationDialectError
-    options:
-      heading_level: 3

pyproject.toml

@@ -82,7 +82,7 @@ testpaths = ["tests"]
 python_files = ["test_*.py"]
 python_classes = ["Test*"]
 python_functions = ["test_*"]
-addopts = "-ra"
+addopts = "-ra -m 'not db_dialect'"
 
 [tool.pyright]
 reportMissingTypeStubs = false

src/orm_loader/backends/__init__.py

@@ -1,13 +1,17 @@
 from .postgres import PostgresBackend
 from .resolve import resolve_backend
 from .sqlite import SQLiteBackend
-from .base import BackendCapabilities, DatabaseBackend, STAGING_SCHEMA, Dialect
+from .base import (
+    BackendCapabilities,
+    DatabaseBackend,
+    Dialect,
+    STAGING_SCHEMA,
+)
 from ..mappers.materialised_view_errors import (
     ConcurrentRefreshNotEligibleError,
     MaterializationError,
     MaterializationFailure,
     MaterializationOperation,
-    UnsupportedMaterializationDialectError,
 )
 
 __all__ = [
@@ -21,6 +25,5 @@
     "PostgresBackend",
     "STAGING_SCHEMA",
     "SQLiteBackend",
-    "UnsupportedMaterializationDialectError",
     "resolve_backend",
 ]

src/orm_loader/backends/base.py

@@ -3,7 +3,6 @@
 from abc import ABC, abstractmethod
 from contextlib import AbstractContextManager, contextmanager, nullcontext
 from dataclasses import dataclass
-from enum import Enum
 from collections.abc import Generator
 from functools import wraps
 from typing import TYPE_CHECKING, Any, Callable, ParamSpec, Type, TypeVar, cast
@@ -13,6 +12,8 @@
 from sqlalchemy.engine import Connection, Engine
 from sqlalchemy.sql.compiler import IdentifierPreparer
 
+from oa_configurator import Dialect, open_connection, qualified
+
 if TYPE_CHECKING:
     from ..loaders.data_classes import LoaderContext
     from ..mappers.materialised_view_contracts import MaterializedViewIndex
@@ -34,13 +35,6 @@ class BackendCapabilities:
     supports_materialized_views: bool = False
 
 
-class Dialect(str, Enum):
-    """Supported SQLAlchemy dialect names."""
-
-    SQLITE = "sqlite"
-    POSTGRESQL = "postgresql"
-
-
 STAGING_SCHEMA: str = "staging"
 
 P = ParamSpec("P")
@@ -126,10 +120,8 @@ def qualified_staging_name(self, tablename: str) -> str:
         str
             e.g. '"staging"."_staging_concept"' or '"_staging_concept"'.
         """
-        from ..helpers.sql import qualify_identifier
-
-        return qualify_identifier(
-            self.staging_name_for_table(tablename), self.staging_schema, self.identifier_preparer
+        return qualified(
+            self.identifier_preparer, self.staging_name_for_table(tablename), physical_schema=self.staging_schema
         )
 
     @property
@@ -192,11 +184,43 @@ def _as_connection(
         self,
         bind: Engine | Connection,
     ) -> Generator[Connection, None, None]:
-        if isinstance(bind, Engine):
-            with bind.begin() as conn:
-                yield conn
-        else:
-            yield bind
+        """
+        Normalize a bind into an open connection, guarding its dialect.
+
+        Every backend method that takes a ``bind`` should route it through
+        this context manager rather than opening a connection itself, so the
+        dialect guard below applies uniformly.
+
+        Parameters
+        ----------
+        bind : Engine or Connection
+            An Engine opens a new connection and transaction scoped to this
+            context manager, committing on a clean exit. A Connection is
+            forwarded as-is; its transaction is owned by the caller, and
+            passing the same Connection into several backend calls groups
+            them into one shared transaction.
+
+        Yields
+        ------
+        Connection
+            An open connection whose dialect matches ``self.dialect``.
+
+        Raises
+        ------
+        TypeError
+            If ``bind``'s dialect does not match ``self.dialect``. Guards
+            against a bind resolved through a different backend being
+            passed directly into a method on this one.
+        """
+        if bind.dialect.name != self.dialect.value:
+            raise TypeError(
+                f"{self.name} backend received a {bind.dialect.name!r} connection; "
+                f"expected {self.dialect.value!r}. The bind passed to this method must "
+                "be the same one (or share the same dialect as) the bind resolve_backend() "
+                "was given."
+            )
+        with open_connection(bind) as conn:
+            yield conn
 
     def _insertable_column_names(
         self,
@@ -268,7 +292,6 @@ def merge_replace(
         self,
         table_cls: Type["CSVTableProtocol"],
         session: so.Session,
-        target_name: str,
         pk_cols: list[str],
         *,
         merge_batch_size: int | None = None,
@@ -280,7 +303,6 @@ def merge_upsert(
         self,
         table_cls: Type["CSVTableProtocol"],
         session: so.Session,
-        target_name: str,
         pk_cols: list[str],
         *,
         merge_batch_size: int | None = None,
@@ -292,7 +314,6 @@ def merge_insert(
         self,
         table_cls: Type["CSVTableProtocol"],
         session: so.Session,
-        target_name: str,
         *,
         merge_batch_size: int | None = None,
     ) -> None:
@@ -351,8 +372,9 @@ def create_materialized_view(
     ) -> None:
         """Create a materialized view for the supplied selectable.
 
-        ``schema`` defaults to ``None``, leaving the target unqualified for
-        the connection's ``search_path`` to resolve.
+        *schema* is the view's already-resolved physical schema (or None for
+        the connection's own default); callers resolve which
+        schema_translate_map key to use before calling.
         """
         raise NotImplementedError(
             f"Backend '{self.name}' has not implemented create_materialized_view()"
@@ -370,6 +392,9 @@ def refresh_materialized_view(
     ) -> None:
         """Refresh a materialized view.
 
+        *schema* is the view's already-resolved physical schema, matching
+        ``create_materialized_view``.
+
         ``declared_indexes`` lets supporting backends validate a concurrent
         refresh request without defining a second catalog-based eligibility
         rule. Other backends may ignore it.
@@ -390,10 +415,12 @@ def drop_materialized_view(
     ) -> None:
         """Drop a materialized view.
 
-        This is deliberately non-abstract: the default implementation
-        requires the capability flag and then raises ``NotImplementedError``.
-        Older third-party backend subclasses need no override to receive a
-        clear error when they do not support this operation.
+        *schema* is the view's already-resolved physical schema, matching
+        ``create_materialized_view``. This is deliberately non-abstract: the
+        default implementation requires the capability flag and then raises
+        ``NotImplementedError``. Older third-party backend subclasses need
+        no override to receive a clear error when they do not support this
+        operation.
         """
         raise NotImplementedError(
             f"Backend '{self.name}' has not implemented drop_materialized_view()"
@@ -411,8 +438,9 @@ def create_materialized_view_index(
     ) -> None:
         """Create an index on a materialized view.
 
-        This is deliberately non-abstract for the same compatibility reason
-        as :meth:`drop_materialized_view`.
+        *schema* is the view's already-resolved physical schema, matching
+        ``create_materialized_view``. This is deliberately non-abstract for
+        the same compatibility reason as :meth:`drop_materialized_view`.
         """
         raise NotImplementedError(
             f"Backend '{self.name}' has not implemented "