Reconcile own indices with pre-existing OHDSI-standard indices

omop_alchemy/backends/__init__.py

@@ -4,9 +4,7 @@
     CONCEPT_NAME_TSVECTOR_COLUMN,
     CONCEPT_SYNONYM_NAME_TSVECTOR_COLUMN,
     FeatureNotSupportedError,
-    FullTextAction,
     FullTextError,
-    FullTextResult,
     FullTextTargetConfig,
     backend_supports,
     require_backend_support,
@@ -22,9 +20,7 @@
     "CONCEPT_NAME_TSVECTOR_COLUMN",
     "CONCEPT_SYNONYM_NAME_TSVECTOR_COLUMN",
     "FeatureNotSupportedError",
-    "FullTextAction",
     "FullTextError",
-    "FullTextResult",
     "FullTextTargetConfig",
     "backend_supports",
     "require_backend_support",

omop_alchemy/backends/base.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from enum import StrEnum
 from typing import TYPE_CHECKING, Any
 import sqlalchemy as sa
 
@@ -24,25 +23,6 @@ class FullTextTargetConfig:
     index_name: str
 
 
-class FullTextAction(StrEnum):
-    INSTALL = "install"
-    POPULATE = "populate"
-    DROP = "drop"
-
-
-@dataclass(frozen=True)
-class FullTextResult:
-    target_name: str
-    table_name: str
-    source_column_name: str
-    vector_column_name: str
-    index_name: str
-    action: FullTextAction
-    status: str
-    detail: str
-    row_count: int | None = None
-
-
 class FullTextError(RuntimeError):
     """Raised when a full-text search maintenance operation fails."""
 

omop_alchemy/maintenance/_cli_utils.py

@@ -5,10 +5,12 @@
 import functools
 import inspect
 from dataclasses import dataclass
+from enum import StrEnum
 from typing import Any, Callable, TypeVar
 
 import sqlalchemy as sa
 import typer
+from orm_loader.backends import STAGING_SCHEMA
 from sqlalchemy.exc import SQLAlchemyError
 
 from .tables import TableScope
@@ -19,6 +21,105 @@
 _F = TypeVar("_F", bound=Callable[..., Any])
 
 
+class ReservedSchema(StrEnum):
+    """Schema names reserved for OMOP_Alchemy/orm-loader internal bookkeeping.
+    A user-configured db_schema may never collide with one of these.
+    """
+
+    STAGING = STAGING_SCHEMA
+    MAINTENANCE = "omop_alchemy_maintenance"
+
+
+def reject_reserved_schema(db_schema: str | None) -> None:
+    """Raise if db_schema collides with a schema name reserved for internal bookkeeping."""
+    if db_schema in set(ReservedSchema):
+        raise RuntimeError(
+            f"db_schema cannot be {db_schema!r}: reserved for OMOP_Alchemy/orm-loader internal use."
+        )
+
+
+class Severity(StrEnum):
+    """Coarse-grained outcome classification shared by every maintenance
+    command's status vocabulary.
+
+    Parameters
+    ----------
+    code : str
+        The severity's string value.
+    style : str
+        Rich color/style name used to render any Status with this severity.
+    """
+
+    def __new__(cls, code: str, style: str):
+        obj = str.__new__(cls, code)
+        obj._value_ = code
+        return obj
+
+    def __init__(self, code: str, style: str):
+        self.style = style
+
+    OK = ("ok", "green")
+    INFO = ("info", "cyan")
+    WARNING = ("warning", "yellow")
+    ERROR = ("error", "red")
+
+
+class Status(StrEnum):
+    """A maintenance command result status, carrying its Severity (and, via
+    Severity.style, its render color).
+
+    Parameters
+    ----------
+    code : str
+        The status's string value (unchanged from the plain strings used
+        before this type existed, so existing string comparisons/dict
+        lookups by value keep working).
+    severity : Severity
+        How severe this status is, and (via severity.style) how it renders.
+    """
+
+    def __new__(cls, code: str, severity: Severity):
+        obj = str.__new__(cls, code)
+        obj._value_ = code
+        return obj
+
+    def __init__(self, code: str, severity: Severity):
+        self.severity = severity
+
+    # -- shared across every dry-run/apply-style domain --
+    PLANNED = ("planned", Severity.INFO)
+    APPLIED = ("applied", Severity.OK)
+    SKIPPED = ("skipped", Severity.WARNING)
+
+    # -- domain-specific "applied" words (backup, index restore/capture,
+    #    sequence reset, vocab load) -- same OK severity as APPLIED, kept as
+    #    distinct words since the specific outcome is worth seeing at a glance --
+    CREATED = ("created", Severity.OK)
+    LOADED = ("loaded", Severity.OK)
+    RESET = ("reset", Severity.OK)
+    RESTORED = ("restored", Severity.OK)
+    CAPTURED = ("captured", Severity.OK)
+    READY = ("ready", Severity.OK)
+    PASSED = ("passed", Severity.OK)
+    MATCHED = ("matched", Severity.OK)
+
+    # -- warnings: something worth a look, but not blocking --
+    WARNING = ("warning", Severity.WARNING)
+    LIMITED = ("limited", Severity.WARNING)
+    DRIFTED = ("drifted", Severity.WARNING)
+
+    # -- informational: an intentionally supported state --
+    RENAMED = ("renamed", Severity.INFO)
+
+    # -- errors/failures --
+    MISSING = ("missing", Severity.ERROR)
+    UNEXPECTED = ("unexpected", Severity.ERROR)
+    MISMATCH = ("mismatch", Severity.ERROR)
+    BLOCKED = ("blocked", Severity.ERROR)
+    UNSUPPORTED = ("unsupported", Severity.ERROR)
+    FAILED = ("failed", Severity.ERROR)
+
+
 @dataclass(frozen=True)
 class _ConnContext:
     """Connection context derived from the oa_configurator resolved resource."""
@@ -53,6 +154,7 @@ def wrapper(**kwargs: Any) -> Any:
             try:
                 from ..config import create_cdm_engine, get_cdm_context
                 pkg_config, resolved = get_cdm_context()
+                reject_reserved_schema(resolved.cdm_schema)
                 engine = create_cdm_engine(resolved)
                 conn = _ConnContext(
                     db_schema=resolved.cdm_schema,
@@ -139,9 +241,9 @@ def handle_error(exc: Exception) -> None:
     raise exc
 
 
-def dry_status(dry_run: bool, applied: str = "applied") -> str:
-    """Return 'planned' when dry_run is True, otherwise the applied label."""
-    return "planned" if dry_run else applied
+def dry_status(dry_run: bool, applied: Status = Status.APPLIED) -> Status:
+    """Return Status.PLANNED when dry_run is True, otherwise the applied status."""
+    return Status.PLANNED if dry_run else applied
 
 
 def dry_label(dry_run: bool, planned: str, applied: str) -> str:

omop_alchemy/maintenance/cli_backup.py

@@ -12,7 +12,7 @@
 import typer
 
 from ..backends import resolve_backend, require_backend_support, backend_support_note
-from ._cli_utils import dry_label, dry_status, omop_command
+from ._cli_utils import Status, dry_label, dry_status, omop_command
 from .ui import (
     console,
     render_backup_result,
@@ -41,7 +41,7 @@ class BackupResult:
 
     file_path: str
     backup_format: BackupFormat
-    status: str
+    status: Status
     detail: str
     database_name: str
     backend: str
@@ -87,7 +87,7 @@ def create_database_backup(
     return BackupResult(
         file_path=str(resolved_output_path),
         backup_format=backup_format,
-        status=dry_status(dry_run, applied="created"),
+        status=dry_status(dry_run, applied=Status.CREATED),
         detail=dry_label(dry_run, "Database backup would be created with pg_dump.", "Database backup created with pg_dump."),
         database_name=database_name,
         backend=engine.dialect.name,

omop_alchemy/maintenance/cli_foreign_keys.py

@@ -8,7 +8,7 @@
 import typer
 
 from ..backends import Backend, resolve_backend, require_backend_support, backend_support_note
-from ._cli_utils import dry_label, dry_status, omop_command
+from ._cli_utils import Status, dry_label, dry_status, omop_command
 from .tables import (
     TableCategory,
     existing_maintenance_tables,
@@ -46,7 +46,7 @@ class _FKTableInfo(ForeignKeyBase):
 class ForeignKeyManagementResult(_FKTableInfo):
     """Outcome of a FK trigger enable or disable operation for one table."""
     enable: bool
-    status: str
+    status: Status
     detail: str
 
 
@@ -62,7 +62,7 @@ class ForeignKeyValidationResult(_FKTableInfo):
     """FK constraint validation outcome for one table, with counts of violating constraints and rows."""
     violating_constraint_count: int
     violating_row_count: int
-    status: str
+    status: Status
     detail: str
 
 
@@ -257,7 +257,7 @@ def validate_foreign_key_constraints(
                 incoming_constraint_count=target.incoming_constraint_count,
                 violating_constraint_count=violating_constraint_count,
                 violating_row_count=violating_row_count,
-                status="failed" if violations else "passed",
+                status=Status.FAILED if violations else Status.PASSED,
                 detail=(
                     _fk_violation_detail(violations)
                     if violations
@@ -314,7 +314,7 @@ def manage_foreign_key_triggers(
                             outgoing_constraint_count=target.outgoing_constraint_count,
                             incoming_constraint_count=target.incoming_constraint_count,
                             enable=enable,
-                            status="failed" if violations else "skipped",
+                            status=Status.FAILED if violations else Status.SKIPPED,
                             detail=(
                                 _fk_violation_detail(violations, strict_abort=True)
                                 if violations

omop_alchemy/maintenance/cli_fulltext.py

@@ -2,13 +2,16 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
+from enum import StrEnum
+
 import typer
 from sqlalchemy.engine import Engine
 
 from ..backends import backend_support_note as _backend_support_note
 from ..backends import resolve_backend, require_backend_support
-from ..backends.base import FullTextAction, FullTextError, FullTextResult
-from ._cli_utils import dry_label, dry_status, omop_command
+from ..backends.base import FullTextError
+from ._cli_utils import Status, dry_label, dry_status, omop_command
 from .ui import (
     console,
     render_fulltext_results,
@@ -21,6 +24,28 @@
 )
 
 
+class FullTextAction(StrEnum):
+    """Which full-text sidecar-column operation a FullTextResult reports on."""
+
+    INSTALL = "install"
+    POPULATE = "populate"
+    DROP = "drop"
+
+
+@dataclass(frozen=True)
+class FullTextResult:
+    """Outcome of a full-text sidecar-column operation for one vocabulary table."""
+    target_name: str
+    table_name: str
+    source_column_name: str
+    vector_column_name: str
+    index_name: str
+    action: FullTextAction
+    status: Status
+    detail: str
+    row_count: int | None = None
+
+
 # ── Orchestrators ─────────────────────────────────────────────────────────────
 
 def install_fulltext_columns(