Display impacts in cli

Author: shangyian

datajunction-clients/python/datajunction/cli.py

@@ -22,7 +22,6 @@
 logger = logging.getLogger(__name__)
 
 
-
 class DJCLI:
     """DJ command-line tool"""
 

datajunction-clients/python/datajunction/deployment.py

@@ -56,7 +56,7 @@
 def _short_name(name: str, namespace: str) -> str:
     """Strip the namespace prefix from a node name for compact display."""
     prefix = namespace + "."
-    return name[len(prefix):] if name.startswith(prefix) else name
+    return name[len(prefix) :] if name.startswith(prefix) else name
 
 
 def _node_type_display(node_type: str) -> str:
@@ -65,11 +65,11 @@ def _node_type_display(node_type: str) -> str:
 
 
 def _impact_annotation(imp: dict, namespace: str, current_parent: str = "") -> str:
-    """Dim annotation: cube reason or multi-cause 'also via' list."""
-    if imp.get("node_type") == "cube":
-        reason = imp.get("impact_reason", "")
-        return f"  [dim]({reason})[/dim]" if reason else ""
+    """Dim annotation: cube 'via' list or multi-cause 'also via' list."""
     caused_by = imp.get("caused_by", [])
+    if imp.get("node_type") == "cube":
+        via = [_short_name(c, namespace) for c in caused_by]
+        return f"  [dim](via: {', '.join(via)})[/dim]" if via else ""
     others = [_short_name(c, namespace) for c in caused_by if c != current_parent]
     return f"  [dim](also via: {', '.join(others)})[/dim]" if others else ""
 
@@ -120,7 +120,7 @@ def _render_impact_item(
     children = [
         ch
         for ch in impacts_by_cause.get(item_name, [])
-        if ch.get("name") not in rendered
+        if ch.get("name") not in rendered and ch.get("node_type") != "cube"
     ]
     rendered.update(ch["name"] for ch in children if ch.get("name"))
     if children:
@@ -150,7 +150,14 @@ def _render_impacts(
         b = indent + ("└" if is_last else "├")
         c = indent + (" " if is_last else "│")
         _render_impact_item(
-            imp, b, c, con, namespace, impacts_by_cause, rendered, current_parent,
+            imp,
+            b,
+            c,
+            con,
+            namespace,
+            impacts_by_cause,
+            rendered,
+            current_parent,
         )
 
 
@@ -192,6 +199,28 @@ def _print_changed_fields(console: Console, change: dict) -> None:
         console.print(f"    [red dim]{truncated}[/red dim]")
 
 
+def _collect_transitive_cubes(
+    node_name: str,
+    impacts_by_cause: dict[str, list[dict]],
+) -> list[dict]:
+    """Collect all cube impacts transitively reachable from node_name, deduped by name."""
+    cubes: dict[str, dict] = {}
+    visited: set[str] = set()
+    queue = [node_name]
+    while queue:
+        current = queue.pop()
+        for imp in impacts_by_cause.get(current, []):
+            name = imp.get("name", "")
+            if name in visited:
+                continue
+            visited.add(name)
+            if imp.get("node_type") == "cube":
+                cubes[name] = imp
+            else:
+                queue.append(name)
+    return list(cubes.values())
+
+
 def _print_change_tree(
     console: Console,
     change: dict,
@@ -201,20 +230,25 @@ def _print_change_tree(
     """Print the tree of column changes, dim link changes, and downstream impacts."""
     column_changes = change.get("column_changes", [])
     dim_link_changes = change.get("dim_link_changes", [])
-    node_downstream = impacts_by_cause.get(change.get("name", ""), [])
+    change_name = change.get("name", "")
+    node_downstream = impacts_by_cause.get(change_name, [])
 
     explicit_removals = [d for d in dim_link_changes if d.get("operation") == "removed"]
     dim_link_additions = [d for d in dim_link_changes if d.get("operation") == "added"]
     dim_link_updates = [d for d in dim_link_changes if d.get("operation") == "updated"]
     broken_by_col = _build_broken_by_col(dim_link_changes)
     rendered: set[str] = set()
 
+    non_cube_downstream = [d for d in node_downstream if d.get("node_type") != "cube"]
+    cube_downstream = _collect_transitive_cubes(change_name, impacts_by_cause)
+
     tree_items = (
         list(column_changes)
         + explicit_removals
         + dim_link_additions
         + dim_link_updates
-        + node_downstream
+        + non_cube_downstream
+        + cube_downstream
     )
     for i, item in enumerate(tree_items):
         is_last = i == len(tree_items) - 1
@@ -257,8 +291,14 @@ def _print_change_tree(
             item_name = item.get("name", "")
             rendered.add(item_name)
             _render_impact_item(
-                item, branch, cont, console, namespace, impacts_by_cause,
-                rendered, change.get("name", ""),
+                item,
+                branch,
+                cont,
+                console,
+                namespace,
+                impacts_by_cause,
+                rendered,
+                change_name,
             )
 
 
@@ -286,13 +326,17 @@ def _print_impact_summary(
             f"[yellow]{update_count} update{'s' if update_count != 1 else ''}[/yellow]",
         )
     if delete_count:
-        summary_parts.append(f"[red]{delete_count} delete{'s' if delete_count != 1 else ''}[/red]")
+        summary_parts.append(
+            f"[red]{delete_count} delete{'s' if delete_count != 1 else ''}[/red]",
+        )
     if skip_count:
         summary_parts.append(f"[dim]{skip_count} skipped[/dim]")
     if will_invalidate:
         summary_parts.append(f"[red]{will_invalidate} downstream invalidated[/red]")
     if may_affect:
-        summary_parts.append(f"[yellow]{may_affect} downstream may be affected[/yellow]")
+        summary_parts.append(
+            f"[yellow]{may_affect} downstream may be affected[/yellow]",
+        )
 
     if not active_changes and not downstream_impacts:
         console.print(
@@ -529,7 +573,6 @@ def pull(
             }
             yaml.safe_dump(project_spec, yaml_file, sort_keys=False)
 
-
     def push(
         self,
         source_path: str | Path,
@@ -596,7 +639,8 @@ def push(
         if deployment_data.get("status") == "failed":
             changes = deployment_data.get("changes", [])
             errors = [
-                c for c in changes
+                c
+                for c in changes
                 if c.get("validation_errors") or c.get("predicted_status") == "invalid"
             ]
             raise DJDeploymentFailure(