fix(deploy): address MilagrosMarin review on #1466

- Replace `assert` statements with explicit `if … raise DataJointError(…)`
  in the dispatch ladder. Project convention is to avoid assertions for
  runtime guarantees because Python's -O strips them; deploy-hook
  environments may run under -O and would otherwise see AttributeError
  on None instead of the intended DataJointError. The unactivated-schema
  case (database is None) is the most concerning since it would have
  emitted a malformed ALTER TABLE.

- Add four unit tests covering previously-untested branches:
  - `test_set_replica_identity_table_instance_target` — Table-instance
    dispatch (deploy.py: isinstance(target, Table) branch).
  - `test_set_replica_identity_table_class_target` — Table-class
    dispatch (deploy.py: issubclass(target, Table) branch).
  - `test_set_replica_identity_case_insensitive_mode` — mode="FULL"
    accepted, matching adapter's case-tolerant handling.
  - `test_set_replica_identity_unactivated_schema_raises` — Schema with
    database=None raises rather than emitting malformed DDL.

- Accept uppercase mode strings (`mode.lower()` normalization) for
  parity with the PG adapter's case-tolerant DDL emission.

- Tighten type hints: `target: TargetType` (Union of Schema/Table-class/
  Table-instance via TYPE_CHECKING forward references), `mode: Literal[
  "default", "full"]`. Runtime validation unchanged.

- Docstring additions:
  - Cost section now flags the brief AccessExclusiveLock the ALTER takes.
  - New "Partial-failure semantics" paragraph explaining that exceptions
    on table N of M leave first N-1 tables modified but propagate without
    returning the partial summary; idempotency makes re-running safe.

- Style: split the adjacent-string concatenation on the "PostgreSQL-only"
  error into a single literal.

All 20 unit tests pass (was 16 + 4 new).

Author: dimitri-yatsenko

src/datajoint/deploy.py

@@ -20,12 +20,22 @@
 
 from __future__ import annotations
 
-from typing import Any
+from typing import TYPE_CHECKING, Any, Literal, Union
 
 from .errors import DataJointError
 
+if TYPE_CHECKING:
+    from .schemas import _Schema
+    from .table import Table
+
+    TargetType = Union["_Schema", type["Table"], "Table"]
 
-def set_replica_identity(target: Any, mode: str = "full", dry_run: bool = True) -> dict:
+
+def set_replica_identity(
+    target: "TargetType",
+    mode: Literal["default", "full"] = "full",
+    dry_run: bool = True,
+) -> dict:
     """
     Apply ``ALTER TABLE ... REPLICA IDENTITY <mode>`` to a schema or table on PostgreSQL.
 
@@ -52,12 +62,24 @@ def set_replica_identity(target: Any, mode: str = "full", dry_run: bool = True)
 
     Cost
     ----
-    The ALTER itself is metadata-only and instant. The cost is in WAL volume
-    after the change: UPDATE/DELETE on tables with FULL log the entire old row,
-    which can be sizable on tables with TOASTed bytea columns. For DataJoint's
-    typical insert-append workload, this cost is negligible. The notable
-    scenario is bulk ``delete()`` on tables with ``<blob>`` columns — a
-    transient WAL burst proportional to the deleted-row payload size.
+    The ALTER itself is metadata-only and instant, but requires a brief
+    ``AccessExclusiveLock`` on each table — it will block behind in-flight
+    writes/reads on a busy table. Run during a quiet window on actively-
+    ingested tables.
+
+    The ongoing cost is in WAL volume after the change: UPDATE/DELETE on
+    tables with FULL log the entire old row, which can be sizable on tables
+    with TOASTed bytea columns. For DataJoint's typical insert-append
+    workload, this cost is negligible. The notable scenario is bulk
+    ``delete()`` on tables with ``<blob>`` columns — a transient WAL burst
+    proportional to the deleted-row payload size.
+
+    Partial-failure semantics
+    -------------------------
+    If ``connection.query(ddl)`` raises on table N of M, the first N-1
+    tables are already modified at the storage layer but the exception
+    propagates without returning the partial summary. The operation is
+    idempotent, so re-running brings the remaining tables into compliance.
 
     Compliance considerations
     -------------------------
@@ -112,35 +134,41 @@ def set_replica_identity(target: Any, mode: str = "full", dry_run: bool = True)
     Databricks: `Lakehouse Sync
     <https://docs.databricks.com/aws/en/oltp/projects/lakehouse-sync>`_.
     """
-    if mode not in ("default", "full"):
+    mode_normalized = mode.lower() if isinstance(mode, str) else mode
+    if mode_normalized not in ("default", "full"):
         raise DataJointError(f"mode must be 'default' or 'full'; got {mode!r}")
+    mode = mode_normalized  # type: ignore[assignment]
 
     from .schemas import _Schema
     from .table import Table
 
     if isinstance(target, _Schema):
         connection = target.connection
-        assert connection is not None, "Schema has no active connection"
+        if connection is None:
+            raise DataJointError("Schema has no active connection.")
         adapter = connection.adapter
-        assert target.database is not None, "Schema is not activated"
+        if target.database is None:
+            raise DataJointError("Schema is not activated. Call schema.activate(...) before set_replica_identity().")
         tables = [adapter.make_full_table_name(target.database, t) for t in target.list_tables()]
     elif isinstance(target, type) and issubclass(target, Table):
         instance = target()
         connection = instance.connection
-        assert connection is not None, "Table has no active connection"
+        if connection is None:
+            raise DataJointError(f"Table {target.__name__} has no active connection.")
         adapter = connection.adapter
         tables = [instance.full_table_name]
     elif isinstance(target, Table):
         connection = target.connection
-        assert connection is not None, "Table has no active connection"
+        if connection is None:
+            raise DataJointError(f"Table {type(target).__name__} has no active connection.")
         adapter = connection.adapter
         tables = [target.full_table_name]
     else:
         raise DataJointError(f"target must be a Schema or Table class/instance; got {type(target).__name__}")
 
     if not hasattr(adapter, "replica_identity_ddl"):
         raise DataJointError(
-            f"set_replica_identity is PostgreSQL-only; the {adapter.backend} adapter " "does not support REPLICA IDENTITY."
+            f"set_replica_identity is PostgreSQL-only; the {adapter.backend} adapter does not support REPLICA IDENTITY."
         )
 
     result: dict[str, Any] = {

tests/unit/test_deploy.py

@@ -107,3 +107,66 @@ def test_set_replica_identity_empty_schema():
     schema = _FakeSchema("ms", [], _FakeAdapter())
     result = set_replica_identity(schema, mode="full", dry_run=False)
     assert result == {"tables_analyzed": 0, "tables_modified": 0, "ddl": []}
+
+
+class _FakeTable(dj.Table):
+    """Table stub bypassing schema-decoration wiring."""
+
+    # Suppress dj.Table's class-construction checks
+    table_name = "fake_table"
+
+    def __init__(self, full_table_name: str, adapter: object) -> None:
+        # Skip dj.Table.__init__ — fabricate the minimal attributes.
+        self._full_table_name = full_table_name
+        self._connection = _FakeConnection(adapter)
+
+    @property
+    def full_table_name(self) -> str:
+        return self._full_table_name
+
+    @property
+    def connection(self):
+        return self._connection
+
+
+def test_set_replica_identity_table_instance_target():
+    """Table instance dispatch (deploy.py: isinstance(target, Table) branch)."""
+    table = _FakeTable('"ms"."the_table"', _FakeAdapter())
+    result = set_replica_identity(table, mode="full", dry_run=False)
+    assert result == {
+        "tables_analyzed": 1,
+        "tables_modified": 1,
+        "ddl": ['ALTER TABLE "ms"."the_table" REPLICA IDENTITY FULL'],
+    }
+    assert table.connection.queries == result["ddl"]
+
+
+def test_set_replica_identity_table_class_target(monkeypatch):
+    """Table-class dispatch (deploy.py: issubclass(target, Table) branch)."""
+    # Build a class that instantiates a _FakeTable when called like target()
+    fake_adapter = _FakeAdapter()
+
+    class _TableClass(dj.Table):
+        def __new__(cls):
+            return _FakeTable('"ms"."class_table"', fake_adapter)
+
+    # `isinstance(_TableClass, type) and issubclass(_TableClass, dj.Table)` is True.
+    result = set_replica_identity(_TableClass, mode="full", dry_run=True)
+    assert result["tables_analyzed"] == 1
+    assert result["tables_modified"] == 0  # dry_run
+    assert result["ddl"] == ['ALTER TABLE "ms"."class_table" REPLICA IDENTITY FULL']
+
+
+def test_set_replica_identity_case_insensitive_mode():
+    """`mode='FULL'` (uppercase) should be accepted, matching adapter case-handling."""
+    schema = _FakeSchema("ms", ["t1"], _FakeAdapter())
+    result = set_replica_identity(schema, mode="FULL", dry_run=True)
+    assert result["ddl"] == ['ALTER TABLE "ms"."t1" REPLICA IDENTITY FULL']
+
+
+def test_set_replica_identity_unactivated_schema_raises():
+    """Schema with database=None (never activated) must raise, not produce malformed DDL."""
+    schema = _FakeSchema("ms", ["t1"], _FakeAdapter())
+    schema.database = None
+    with pytest.raises(DataJointError, match="Schema is not activated"):
+        set_replica_identity(schema, mode="full", dry_run=True)