Coverage for python/pyairflowtester/cli.py: 0%

1""" 

2PyAirflowTester CLI: Command-line interface for artifact and runtime analysis. 

3""" 

4 

5import logging 

6from pathlib import Path 

7 

8import click 

9from rich.console import Console 

10from rich.table import Table 

11 

12from pyairflowtester.analyzer import Analyzer, AnalyzerNotImplementedError 

13from pyairflowtester.report import ReportGenerator 

14from pyairflowtester.scanner import Scanner 

15 

16# Configure logging 

17logging.basicConfig( 

18 level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" 

19) 

20logger = logging.getLogger(__name__) 

21 

22console = Console() 

23 

24 

25@click.group() 

26@click.version_option() 

27def main(): 

28 """PyAirflowTester: Unified Airflow + dbt Reliability Platform""" 

29 pass 

30 

31 

32@main.command() 

33@click.argument("path", type=click.Path(exists=True), default=".") 

34@click.option( 

35 "--dags", 

36 type=click.Path(exists=True), 

37 help="Path to Airflow DAGs directory", 

38) 

39@click.option( 

40 "--dbt", 

41 type=click.Path(exists=True), 

42 help="Path to dbt project directory", 

43) 

44@click.option( 

45 "--airflow-cfg", 

46 type=click.Path(exists=True), 

47 help="Path to airflow.cfg for configuration auditing", 

48) 

49@click.option( 

50 "--format", 

51 type=click.Choice(["json", "html", "markdown", "sarif"]), 

52 default="json", 

53 help="Output format", 

54) 

55@click.option( 

56 "--output", 

57 type=click.Path(), 

58 default=None, 

59 help="Output file path", 

60) 

61@click.option( 

62 "--severity", 

63 type=click.Choice(["critical", "high", "medium", "low", "info"]), 

64 default="medium", 

65 help="Minimum severity level to report", 

66) 

67def scan(path, dags, dbt, airflow_cfg, format, output, severity): 

68 """ 

69 Scan Airflow DAGs and dbt projects for issues. 

70 

71 Performs static analysis to detect configuration issues, anti-patterns, 

72 and potential reliability risks before deployment. 

73 """ 

74 console.print("[bold cyan]PyAirflowTester Scanner[/bold cyan]") 

75 console.print(f"Target: {path}\n") 

76 

77 scanner = Scanner() 

78 

79 # Scan DAGs if path provided 

80 if dags: 

81 console.print("[yellow]Scanning Airflow DAGs...[/yellow]") 

82 dag_violations = scanner.scan_dags(Path(dags)) 

83 console.print(f" Found {len(dag_violations)} violations\n") 

84 else: 

85 dag_violations = [] 

86 

87 # Scan dbt if path provided 

88 if dbt: 

89 console.print("[yellow]Scanning dbt project...[/yellow]") 

90 dbt_violations = scanner.scan_dbt(Path(dbt)) 

91 console.print(f" Found {len(dbt_violations)} violations\n") 

92 else: 

93 dbt_violations = [] 

94 

95 # Scan airflow.cfg if path provided 

96 if airflow_cfg: 

97 console.print("[yellow]Scanning Airflow configuration...[/yellow]") 

98 config_violations = scanner.scan_config(Path(airflow_cfg)) 

99 console.print(f" Found {len(config_violations)} violations\n") 

100 else: 

101 config_violations = [] 

102 

103 all_violations = dag_violations + dbt_violations + config_violations 

104 

105 if not all_violations: 

106 console.print("[green]✓ No violations found![/green]") 

107 return 

108 

109 # Filter by severity 

110 from pyairflowtester.models import SEVERITY_WEIGHTS 

111 

112 min_weight = SEVERITY_WEIGHTS.get(severity, 0) 

113 filtered = [ 

114 v 

115 for v in all_violations 

116 if SEVERITY_WEIGHTS.get(v.get("severity", "info"), 0) >= min_weight 

117 ] 

118 

119 # Display results 

120 _display_violations(filtered) 

121 

122 # Generate report if format specified 

123 if format and output: 

124 generator = ReportGenerator() 

125 report_path = generator.generate(format, filtered, Path(output)) 

126 console.print(f"\n[green]Report generated:[/green] {report_path}") 

127 

128 

129@main.command() 

130@click.argument("path", type=click.Path(exists=True), default=".") 

131@click.option( 

132 "--compare", 

133 type=str, 

134 default=None, 

135 help="Compare to baseline (git branch)", 

136) 

137@click.option( 

138 "--format", 

139 type=click.Choice(["json", "markdown", "html"]), 

140 default="markdown", 

141 help="Output format", 

142) 

143def score(path, compare, format): 

144 """ 

145 Calculate risk scores for DAGs and dbt models. 

146 

147 Generates multi-dimensional risk scores based on: 

148 - Failure history and patterns 

149 - Configuration quality 

150 - Test coverage 

151 - Performance metrics 

152 """ 

153 console.print("[bold cyan]PyAirflowTester Scoring[/bold cyan]") 

154 console.print(f"Target: {path}\n") 

155 

156 scanner = Scanner() 

157 scorer = scanner._get_scorer() 

158 

159 # Scan for violations 

160 dags_dir = Path(path) / "dags" 

161 dag_violations = scanner.scan_dags(dags_dir if dags_dir.exists() else Path(path)) 

162 dbt_dir = Path(path) / "dbt" 

163 dbt_violations = scanner.scan_dbt(dbt_dir if dbt_dir.exists() else Path(path)) 

164 

165 all_violations = dag_violations + dbt_violations 

166 

167 # Calculate score 

168 risk_score = scorer.calculate_risk_score(all_violations) 

169 risk_level = scorer.categorize_risk(risk_score) 

170 

171 # Display scorecard 

172 table = Table(title="Risk Scorecard") 

173 table.add_column("Metric", style="cyan") 

174 table.add_column("Score", style="magenta") 

175 table.add_row("Overall Risk", f"{risk_score:.1f}/100") 

176 table.add_row("Risk Level", risk_level) 

177 table.add_row("Total Violations", str(len(all_violations))) 

178 

179 console.print(table) 

180 

181 if compare: 

182 console.print(f"\n[yellow]Comparing to baseline: {compare}[/yellow]") 

183 

184 

185@main.command() 

186@click.option("--category", type=str, default=None, help="Filter by category") 

187@click.option("--severity", type=str, default=None, help="Filter by severity") 

188@click.option("--execution-mode", type=str, default=None, help="Filter by execution mode") 

189def rules(category, severity, execution_mode): 

190 """ 

191 List all available rules. 

192 

193 Rules are organized by category (reliability, performance, etc.) and 

194 execution mode (static, runtime, correlation). 

195 """ 

196 console.print("[bold cyan]PyAirflowTester Rules[/bold cyan]\n") 

197 

198 from pyairflowtester.rules import get_all_rules 

199 

200 all_rules = get_all_rules() 

201 

202 # Filter 

203 filtered_rules = all_rules 

204 if category: 

205 filtered_rules = [r for r in filtered_rules if r.get("category") == category] 

206 if severity: 

207 filtered_rules = [r for r in filtered_rules if r.get("severity") == severity] 

208 if execution_mode: 

209 filtered_rules = [r for r in filtered_rules if r.get("execution_mode") == execution_mode] 

210 

211 # Display 

212 table = Table(title=f"Available Rules ({len(filtered_rules)})") 

213 table.add_column("ID", style="cyan") 

214 table.add_column("Name", style="green") 

215 table.add_column("Severity", style="magenta") 

216 table.add_column("Category", style="yellow") 

217 table.add_column("Mode") 

218 

219 for rule in filtered_rules: 

220 table.add_row( 

221 rule.get("id", ""), 

222 rule.get("name", ""), 

223 rule.get("severity", ""), 

224 rule.get("category", ""), 

225 rule.get("execution_mode", ""), 

226 ) 

227 

228 console.print(table) 

229 

230 

231@main.command() 

232@click.option( 

233 "--airflow-home", 

234 type=click.Path(exists=True), 

235 help="Airflow home directory", 

236) 

237@click.option( 

238 "--airflow-db", 

239 type=str, 

240 help="Airflow database connection string", 

241) 

242def connect(airflow_home, airflow_db): 

243 """ 

244 Connect to Airflow instance for runtime analysis. 

245 

246 NOTE: Runtime analysis (the Analyzer subsystem) is not yet implemented. 

247 This command reports that plainly instead of a fake success message. 

248 """ 

249 console.print("[bold cyan]PyAirflowTester Runtime Connection[/bold cyan]\n") 

250 

251 analyzer = Analyzer(airflow_home=airflow_home, airflow_db=airflow_db) 

252 try: 

253 analyzer.connect() 

254 except AnalyzerNotImplementedError as e: 

255 console.print(f"[red]✗ {e}[/red]") 

256 raise SystemExit(1) 

257 

258 

259@main.command() 

260@click.option( 

261 "--dags", 

262 type=click.Path(exists=True), 

263 help="Path to Airflow DAGs directory (same as `pyairflowtester dependency build --dags`)", 

264) 

265@click.option( 

266 "--dbt-manifest", 

267 type=click.Path(exists=True), 

268 help="Path to dbt manifest.json", 

269) 

270@click.option("--host", type=str, default="127.0.0.1", help="Host to bind the server to") 

271@click.option("--port", type=int, default=8080, help="Port to bind the server to") 

272def serve(dags, dbt_manifest, host, port): 

273 """ 

274 Launch the web dashboard. 

275 

276 Builds the dependency graph from --dags/--dbt-manifest (same sources as 

277 `pyairflowtester dependency build`) and serves DashboardBuilder output as 

278 a browsable HTML dashboard at http://HOST:PORT/. 

279 

280 Requires the optional `web` extra: 

281 

282 pip install pyairflowtester[web] 

283 """ 

284 try: 

285 import uvicorn 

286 except ImportError: 

287 console.print("[red]✗ The web dashboard requires optional dependencies.[/red]") 

288 console.print("Install them with: [bold]pip install pyairflowtester\\[web][/bold]") 

289 raise SystemExit(1) 

290 

291 from pyairflowtester.web.app import build_app_from_sources 

292 

293 console.print("[bold cyan]PyAirflowTester Web Dashboard[/bold cyan]") 

294 console.print( 

295 f"Building dependency graph (dags={dags or '-'}, dbt_manifest={dbt_manifest or '-'})\n" 

296 ) 

297 

298 app = build_app_from_sources(dags=dags, dbt_manifest=dbt_manifest) 

299 node_count = len(app.state.graph.nodes) 

300 console.print(f"[green]Graph ready:[/green] {node_count} node(s)") 

301 console.print(f"Serving at [green]http://{host}:{port}[/green] (Ctrl+C to stop)\n") 

302 

303 uvicorn.run(app, host=host, port=port) 

304 

305 

306def _display_violations(violations): 

307 """Display violations in rich table format.""" 

308 if not violations: 

309 return 

310 

311 table = Table(title=f"Violations Found ({len(violations)})") 

312 table.add_column("Rule", style="cyan") 

313 table.add_column("Severity", style="magenta") 

314 table.add_column("Resource", style="green") 

315 table.add_column("Message") 

316 

317 for v in violations[:20]: # Show first 20 

318 severity = v.get("severity", "info") 

319 severity_style = { 

320 "critical": "red", 

321 "high": "red", 

322 "medium": "yellow", 

323 "low": "blue", 

324 "info": "white", 

325 }.get(severity, "white") 

326 

327 table.add_row( 

328 v.get("rule_id", ""), 

329 f"[{severity_style}]{severity}[/{severity_style}]", 

330 v.get("affected_resource", ""), 

331 v.get("message", "")[:50], 

332 ) 

333 

334 if len(violations) > 20: 

335 table.add_row("...", "...", "...", f"... and {len(violations) - 20} more") 

336 

337 console.print(table) 

338 

339 

340# Register the dependency-intelligence command group (graph build/impact/ 

341# blast-radius/etc.). This was previously defined but never attached to the 

342# main CLI group, so `pyairflowtester dependency ...` did not work. 

343from pyairflowtester.dependency_intelligence.cli import register_dependency_cli # noqa: E402 

344 

345register_dependency_cli(main) 

346 

347 

348if __name__ == "__main__": 

349 main()