refactor: make Diagram.cascade() a classmethod factory

The cascade preview pattern is now a single call:

    dj.Diagram.cascade(Session & 'subject_id=1').counts()

cascade() constructs the Diagram directly from the table
expression, includes all descendants (cross-schema), propagates
restrictions, and trims to the affected subgraph.

Table.delete() uses Diagram.cascade(self, ...) internally.
Table.drop() expands descendants inline via nx.descendants().

Removes _from_table() — no longer needed. Also removes dry_run
from delete() and drop() since safemode's transaction + rollback
provides a safer preview, and Diagram.cascade().counts() provides
programmatic preview.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: dimitri-yatsenko

src/datajoint/diagram.py

@@ -100,6 +100,7 @@ def __init__(self, source, context=None) -> None:
             self._restrict_conditions = copy_module.deepcopy(source._restrict_conditions)
             self._restriction_attrs = copy_module.deepcopy(source._restriction_attrs)
             self._part_integrity = source._part_integrity
+            self._source = getattr(source, "_source", None)
             super().__init__(source)
             return
 
@@ -131,6 +132,7 @@ def __init__(self, source, context=None) -> None:
 
         # Enumerate nodes from all the items in the list
         self.nodes_to_show = set()
+        self._source = None
         try:
             self.nodes_to_show.add(source.full_table_name)
         except AttributeError:
@@ -165,39 +167,6 @@ def from_sequence(cls, sequence) -> "Diagram":
         """
         return functools.reduce(lambda x, y: x + y, map(Diagram, sequence))
 
-    @classmethod
-    def _from_table(cls, table_expr) -> "Diagram":
-        """
-        Create a Diagram containing table_expr and all its descendants.
-
-        Internal factory for ``Table.delete()`` and ``Table.drop()``.
-        Bypasses the normal ``__init__`` which does caller-frame introspection
-        and source-type resolution.
-
-        Parameters
-        ----------
-        table_expr : Table
-            A table instance with ``connection`` and ``full_table_name``.
-
-        Returns
-        -------
-        Diagram
-        """
-        conn = table_expr.connection
-        conn.dependencies.load()
-        descendants = set(conn.dependencies.descendants(table_expr.full_table_name))
-        result = cls.__new__(cls)
-        nx.DiGraph.__init__(result, conn.dependencies)
-        result._connection = conn
-        result.context = {}
-        result.nodes_to_show = descendants
-        result._expanded_nodes = set(descendants)
-        result._cascade_restrictions = {}
-        result._restrict_conditions = {}
-        result._restriction_attrs = {}
-        result._part_integrity = "enforce"
-        return result
-
     def add_parts(self) -> "Diagram":
         """
         Add part tables of all masters already in the diagram.
@@ -347,45 +316,65 @@ def topo_sort(self) -> list[str]:
         """
         return topo_sort(self)
 
-    def cascade(self, table_expr, part_integrity="enforce"):
+    @classmethod
+    def cascade(cls, table_expr, part_integrity="enforce"):
         """
-        Apply cascade restriction and propagate downstream.
-
-        OR at convergence — a child row is affected if *any* restricted
-        ancestor taints it. Used for delete.
+        Create a cascade diagram for a table expression.
 
-        Can only be called once on an unrestricted Diagram. Cannot be
-        mixed with ``restrict()``.
+        Builds a Diagram from the table's dependency graph, includes all
+        descendants (across all loaded schemas), and propagates the
+        restriction downstream using OR convergence — a child row is
+        affected if *any* restricted ancestor taints it.
 
         Parameters
         ----------
         table_expr : QueryExpression
-            A restricted table expression
+            A (possibly restricted) table expression
             (e.g., ``Session & 'subject_id=1'``).
         part_integrity : str, optional
             ``"enforce"`` (default), ``"ignore"``, or ``"cascade"``.
 
         Returns
         -------
         Diagram
-            New Diagram with cascade restrictions applied.
+            New Diagram with cascade restrictions applied, trimmed to
+            the seed table and its affected descendants.
+
+        Examples
+        --------
+        >>> # Preview cascade impact across all downstream schemas
+        >>> dj.Diagram.cascade(Session & 'subject_id=1').counts()
+
+        >>> # Inspect the cascade subgraph
+        >>> dj.Diagram.cascade(Session & 'subject_id=1')
         """
-        if self._cascade_restrictions or self._restrict_conditions:
-            raise DataJointError(
-                "cascade() can only be called once on an unrestricted Diagram. "
-                "cascade and restrict modes are mutually exclusive."
-            )
-        result = Diagram(self)
-        result._part_integrity = part_integrity
+        conn = table_expr.connection
+        conn.dependencies.load()
         node = table_expr.full_table_name
-        if node not in result.nodes():
-            raise DataJointError(f"Table {node} is not in the diagram.")
+
+        result = cls.__new__(cls)
+        nx.DiGraph.__init__(result, conn.dependencies)
+        result._connection = conn
+        result.context = {}
+        result._cascade_restrictions = {}
+        result._restrict_conditions = {}
+        result._restriction_attrs = {}
+        result._part_integrity = part_integrity
+        result._source = table_expr
+
+        # Include seed + all descendants
+        descendants = set(nx.descendants(result, node)) | {node}
+        result.nodes_to_show = descendants
+        result._expanded_nodes = set(descendants)
+
         # Seed restriction
         restriction = AndList(table_expr.restriction)
         result._cascade_restrictions[node] = [restriction] if restriction else []
         result._restriction_attrs[node] = set(table_expr.restriction_attributes)
+
         # Propagate downstream
         result._propagate_restrictions(node, mode="cascade", part_integrity=part_integrity)
+
         # Trim graph to cascade subgraph: only restricted tables
         # (seed + descendants) plus alias nodes connecting them.
         keep = set(result._cascade_restrictions)

src/datajoint/table.py

@@ -996,8 +996,7 @@ def delete(
 
         To preview cascade impact without executing, use ``Diagram``::
 
-            diag = dj.Diagram(schema)
-            diag.cascade(MyTable & restriction).counts()
+            dj.Diagram.cascade(MyTable & restriction).counts()
 
         Args:
             transaction: If `True`, use of the entire delete becomes an atomic transaction.
@@ -1022,8 +1021,7 @@ def delete(
             raise ValueError(f"part_integrity must be 'enforce', 'ignore', or 'cascade', " f"got {part_integrity!r}")
         from .diagram import Diagram
 
-        diagram = Diagram._from_table(self)
-        diagram = diagram.cascade(self, part_integrity=part_integrity)
+        diagram = Diagram.cascade(self, part_integrity=part_integrity)
 
         conn = self.connection
         prompt = conn._config["safemode"] if prompt is None else prompt
@@ -1169,9 +1167,14 @@ def drop(self, prompt: bool | None = None, part_integrity: str = "enforce"):
             raise DataJointError(
                 "A table with an applied restriction cannot be dropped. " "Call drop() on the unrestricted Table."
             )
+        import networkx as nx
         from .diagram import Diagram
 
-        diagram = Diagram._from_table(self)
+        diagram = Diagram(self)
+        # Expand to include all descendants (cross-schema)
+        descendants = set(nx.descendants(diagram, self.full_table_name)) | {self.full_table_name}
+        diagram.nodes_to_show = descendants
+        diagram._expanded_nodes = set(descendants)
         conn = self.connection
         prompt = conn._config["safemode"] if prompt is None else prompt
 

tests/integration/test_cascade_delete.py

@@ -217,8 +217,7 @@ class Child(dj.Manual):
     Child.insert1((2, 1, "C2-1"))
 
     # Preview restricted cascade via Diagram
-    diag = dj.Diagram._from_table(Parent & {"parent_id": 1})
-    counts = diag.cascade(Parent & {"parent_id": 1}).counts()
+    counts = dj.Diagram.cascade(Parent & {"parent_id": 1}).counts()
 
     assert isinstance(counts, dict)
     assert counts[Parent.full_table_name] == 1

tests/integration/test_erd.py

@@ -126,8 +126,7 @@ def test_prune_after_restrict(schema_simp_pop):
 
 def test_prune_after_cascade(schema_simp_pop):
     """Prune after cascade removes tables with zero matching rows."""
-    diag = dj.Diagram(schema_simp_pop, context=LOCALS_SIMPLE)
-    cascaded = diag.cascade(A & "id_a=0")
+    cascaded = dj.Diagram.cascade(A & "id_a=0")
     counts = cascaded.counts()
 
     pruned = cascaded.prune()