Coverage for python/pyairflowtester/dependency_intelligence/cli.py: 0%
167 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 20:43 +0530
1"""CLI commands for dependency intelligence."""
3import json
4import logging
5from pathlib import Path
6from typing import Optional
8import click
9from rich.console import Console
10from rich.panel import Panel
11from rich.table import Table
13from .analyzers import (
14 BlastRadiusEngine,
15 ImpactAnalysisEngine,
16 RiskScoringEngine,
17)
18from .graph import DependencyGraphEngine
19from .models import NodeSeverity
20from .parsers import UnifiedGraphBuilder
22console = Console()
23logger = logging.getLogger(__name__)
26@click.group()
27def dependency_cli():
28 """Dependency Intelligence commands."""
29 pass
32@dependency_cli.command()
33@click.option("--dags", type=click.Path(), help="Path to DAGs directory")
34@click.option("--dbt-manifest", type=click.Path(), help="Path to dbt manifest.json")
35@click.option("--datasets", type=click.Path(), help="Path to datasets directory")
36@click.option("--output", type=click.File("w"), default="-", help="Output file")
37def build(dags: Optional[str], dbt_manifest: Optional[str], datasets: Optional[str], output):
38 """Build dependency graph from sources."""
39 console.print("[bold]Building dependency graph...[/bold]")
41 # Collect DAG files
42 dag_files = []
43 if dags:
44 dag_path = Path(dags)
45 dag_files = list(dag_path.glob("**/*.py"))
47 # Build unified graph
48 graph = UnifiedGraphBuilder.build_unified_graph(
49 dag_files=[str(f) for f in dag_files],
50 dbt_manifest=dbt_manifest,
51 dataset_files=[],
52 )
54 # Display stats
55 stats = graph.stats()
56 console.print(
57 Panel(
58 f"[bold]Graph built successfully[/bold]\n"
59 f"Nodes: {stats['node_count']}\n"
60 f"Edges: {stats['edge_count']}\n"
61 f"Critical: {stats['critical_nodes']}",
62 title="Graph Statistics",
63 )
64 )
66 # Output graph. Enum instances (e.g. node_types keys/values in stats)
67 # aren't JSON-serializable as-is, so normalize them to strings first.
68 json_safe_stats = dict(stats)
69 if "node_types" in json_safe_stats:
70 json_safe_stats["node_types"] = {
71 node_type.value: count for node_type, count in json_safe_stats["node_types"].items()
72 }
74 output.write(
75 json.dumps(
76 {
77 "nodes": {
78 k: {
79 "id": v.id,
80 "name": v.name,
81 "type": v.type.value,
82 "owner": v.owner,
83 "severity": v.severity.value,
84 }
85 for k, v in graph.nodes.items()
86 },
87 "edges": [
88 {
89 "source": e.source,
90 "target": e.target,
91 "relationship_type": e.relationship_type.value,
92 }
93 for e in graph.edges
94 ],
95 "stats": json_safe_stats,
96 },
97 indent=2,
98 )
99 )
102@dependency_cli.command()
103@click.argument("node_id")
104@click.option("--depth", type=int, default=None, help="Max traversal depth")
105@click.option("--dags", type=click.Path(), help="Path to DAGs directory")
106@click.option("--dbt-manifest", type=click.Path(), help="Path to dbt manifest.json")
107def impact(node_id: str, depth: Optional[int], dags: Optional[str], dbt_manifest: Optional[str]):
108 """Analyze impact of changing a node."""
109 # Build graph
110 dag_files = []
111 if dags:
112 dag_path = Path(dags)
113 dag_files = list(dag_path.glob("**/*.py"))
115 graph = UnifiedGraphBuilder.build_unified_graph(
116 dag_files=[str(f) for f in dag_files],
117 dbt_manifest=dbt_manifest,
118 )
120 # Run impact analysis
121 engine = ImpactAnalysisEngine(graph)
122 result = engine.analyze(node_id, max_depth=depth)
124 # Display results
125 console.print(
126 Panel(
127 f"[bold]Impact Analysis: {node_id}[/bold]\n"
128 f"Impacted Nodes: {result.impact_depth}\n"
129 f"Impact Score: {result.impact_score:.2f}\n"
130 f"Affected: {len(result.impacted_nodes)}",
131 title="Impact Analysis",
132 )
133 )
135 # Show affected nodes by severity
136 if result.by_severity:
137 console.print("\n[bold]By Severity:[/bold]")
138 severities = [
139 NodeSeverity.CRITICAL,
140 NodeSeverity.HIGH,
141 NodeSeverity.MEDIUM,
142 NodeSeverity.LOW,
143 ]
144 for severity in severities:
145 nodes = result.by_severity.get(severity, [])
146 if nodes:
147 console.print(f" {severity.value.upper()}: {len(nodes)}")
150@dependency_cli.command()
151@click.option("--dags", type=click.Path(), help="Path to DAGs directory")
152@click.option("--dbt-manifest", type=click.Path(), help="Path to dbt manifest.json")
153@click.option("--format", type=click.Choice(["text", "mermaid", "graphviz"]), default="text")
154def lineage(dags: Optional[str], dbt_manifest: Optional[str], format: str):
155 """Show dependency lineage."""
156 # Build graph
157 dag_files = []
158 if dags:
159 dag_path = Path(dags)
160 dag_files = list(dag_path.glob("**/*.py"))
162 graph = UnifiedGraphBuilder.build_unified_graph(
163 dag_files=[str(f) for f in dag_files],
164 dbt_manifest=dbt_manifest,
165 )
167 if format == "mermaid":
168 # Output Mermaid diagram
169 console.print("```mermaid")
170 console.print("graph TD")
171 for edge in graph.edges[:50]: # Limit to first 50 edges
172 source = graph.nodes.get(edge.source, None)
173 target = graph.nodes.get(edge.target, None)
174 if source and target:
175 console.print(f" {source.name} --> {target.name}")
176 console.print("```")
177 else:
178 # Text format
179 table = Table(title="Dependencies")
180 table.add_column("Source", style="cyan")
181 table.add_column("Target", style="magenta")
182 table.add_column("Type", style="green")
184 for edge in graph.edges[:20]: # Show first 20
185 source = graph.nodes.get(edge.source, None)
186 target = graph.nodes.get(edge.target, None)
187 if source and target:
188 table.add_row(source.name, target.name, edge.relationship_type.value)
190 console.print(table)
193@dependency_cli.command()
194@click.option("--nodes", "-n", multiple=True, required=True, help="Nodes that changed")
195@click.option("--dags", type=click.Path(), help="Path to DAGs directory")
196@click.option("--dbt-manifest", type=click.Path(), help="Path to dbt manifest.json")
197def blast_radius(nodes: tuple, dags: Optional[str], dbt_manifest: Optional[str]):
198 """Analyze blast radius of changes."""
199 # Build graph
200 dag_files = []
201 if dags:
202 dag_path = Path(dags)
203 dag_files = list(dag_path.glob("**/*.py"))
205 graph = UnifiedGraphBuilder.build_unified_graph(
206 dag_files=[str(f) for f in dag_files],
207 dbt_manifest=dbt_manifest,
208 )
210 # Run blast radius analysis
211 engine = BlastRadiusEngine(graph)
212 result = engine.analyze(list(nodes))
214 # Display results
215 console.print(
216 Panel(
217 f"[bold]Blast Radius Analysis[/bold]\n"
218 f"Changed Nodes: {len(result.change_nodes)}\n"
219 f"Affected Nodes: {result.blast_radius}\n"
220 f"Blast Depth: {result.blast_depth}\n"
221 f"Risk Level: {result.risk_level.upper()}\n"
222 f"Deployable: {'✓' if result.deployable else '✗'}",
223 title="Blast Radius",
224 border_style="red" if not result.deployable else "green",
225 )
226 )
228 # Show severity distribution
229 if result.severity_distribution:
230 console.print("\n[bold]Severity Distribution:[/bold]")
231 for severity, count in sorted(
232 result.severity_distribution.items(), key=lambda x: x[0].value, reverse=True
233 ):
234 console.print(f" {severity.value.upper()}: {count}")
237@dependency_cli.command()
238@click.option("--dags", type=click.Path(), help="Path to DAGs directory")
239@click.option("--dbt-manifest", type=click.Path(), help="Path to dbt manifest.json")
240def detect_cycles(dags: Optional[str], dbt_manifest: Optional[str]):
241 """Detect circular dependencies."""
242 # Build graph
243 dag_files = []
244 if dags:
245 dag_path = Path(dags)
246 dag_files = list(dag_path.glob("**/*.py"))
248 graph = UnifiedGraphBuilder.build_unified_graph(
249 dag_files=[str(f) for f in dag_files],
250 dbt_manifest=dbt_manifest,
251 )
253 # Run cycle detection
254 engine = DependencyGraphEngine(graph)
255 cycles = engine.detect_cycles()
257 if cycles:
258 console.print(
259 Panel(
260 f"[bold red]⚠ Found {len(cycles)} circular dependencies[/bold red]",
261 title="Cycle Detection",
262 )
263 )
265 table = Table(title="Cycles")
266 table.add_column("Cycle #", style="cyan")
267 table.add_column("Path", style="magenta")
269 for i, cycle in enumerate(cycles, 1):
270 path_str = " → ".join(graph.nodes[n].name if n in graph.nodes else n for n in cycle[:5])
271 if len(cycle) > 5:
272 path_str += f" ... (+{len(cycle) - 5} more)"
273 table.add_row(str(i), path_str)
275 console.print(table)
276 else:
277 console.print("[bold green]✓ No circular dependencies detected[/bold green]")
280@dependency_cli.command()
281@click.option("--dags", type=click.Path(), help="Path to DAGs directory")
282@click.option("--dbt-manifest", type=click.Path(), help="Path to dbt manifest.json")
283def detect_orphans(dags: Optional[str], dbt_manifest: Optional[str]):
284 """Detect orphaned nodes."""
285 # Build graph
286 dag_files = []
287 if dags:
288 dag_path = Path(dags)
289 dag_files = list(dag_path.glob("**/*.py"))
291 graph = UnifiedGraphBuilder.build_unified_graph(
292 dag_files=[str(f) for f in dag_files],
293 dbt_manifest=dbt_manifest,
294 )
296 # Run orphan detection
297 engine = DependencyGraphEngine(graph)
298 orphans = engine.detect_orphans()
300 # Display results
301 console.print(
302 Panel(
303 f"[bold]Orphan Detection Results[/bold]\n"
304 f"Sources (no incoming): {len(orphans['sources'])}\n"
305 f"Sinks (no outgoing): {len(orphans['sinks'])}\n"
306 f"Isolated: {len(orphans['isolated'])}",
307 title="Orphan Detection",
308 )
309 )
311 if orphans["isolated"]:
312 console.print("\n[bold yellow]Isolated Nodes:[/bold yellow]")
313 for node_id in orphans["isolated"][:10]:
314 if node_id in graph.nodes:
315 console.print(f" - {graph.nodes[node_id].name}")
318@dependency_cli.command()
319@click.option("--top", type=int, default=10, help="Show top N nodes")
320@click.option("--dags", type=click.Path(), help="Path to DAGs directory")
321@click.option("--dbt-manifest", type=click.Path(), help="Path to dbt manifest.json")
322def risk_score(top: int, dags: Optional[str], dbt_manifest: Optional[str]):
323 """Calculate risk scores for all nodes."""
324 # Build graph
325 dag_files = []
326 if dags:
327 dag_path = Path(dags)
328 dag_files = list(dag_path.glob("**/*.py"))
330 graph = UnifiedGraphBuilder.build_unified_graph(
331 dag_files=[str(f) for f in dag_files],
332 dbt_manifest=dbt_manifest,
333 )
335 # Run risk scoring
336 engine = RiskScoringEngine(graph)
337 scores = engine.score_all_nodes()
339 # Sort by risk score
340 sorted_scores = sorted(scores.items(), key=lambda x: x[1].risk_score, reverse=True)
342 # Display top N
343 table = Table(title=f"Top {top} Highest Risk Nodes")
344 table.add_column("Node", style="cyan")
345 table.add_column("Risk Score", style="red")
346 table.add_column("Severity", style="yellow")
347 table.add_column("Downstream", style="green")
349 for node_id, result in sorted_scores[:top]:
350 node = graph.nodes.get(node_id)
351 if node:
352 table.add_row(
353 node.name,
354 f"{result.risk_score:.1f}",
355 result.severity.value.upper(),
356 str(result.metadata.get("downstream_nodes", 0)),
357 )
359 console.print(table)
362def register_dependency_cli(main_cli):
363 """Register dependency CLI group with main CLI."""
364 main_cli.add_command(dependency_cli, name="dependency")