Coverage for python/pyairflowtester/dependency_intelligence/analyzers.py: 88%

1"""Analysis engines for dependency intelligence.""" 

2 

3import logging 

4from datetime import datetime 

5from typing import Dict, List, Optional 

6 

7from .graph import DependencyGraphEngine 

8from .models import ( 

9 BlastRadiusResult, 

10 DependencyGraph, 

11 DriftDetectionResult, 

12 ImpactResult, 

13 Node, 

14 NodeSeverity, 

15 NodeType, 

16 RiskScoreResult, 

17) 

18 

19logger = logging.getLogger(__name__) 

20 

21 

22class ImpactAnalysisEngine: 

23 """Analyze impact of changing a node on downstream dependencies.""" 

24 

25 def __init__(self, graph: DependencyGraph): 

26 self.graph = graph 

27 self.engine = DependencyGraphEngine(graph) 

28 

29 def analyze(self, node_id: str, max_depth: Optional[int] = None) -> ImpactResult: 

30 """ 

31 Analyze impact of changing a node. 

32 

33 Args: 

34 node_id: Node to analyze 

35 max_depth: Maximum traversal depth 

36 

37 Returns: 

38 ImpactResult with affected nodes and metrics 

39 """ 

40 if node_id not in self.graph.nodes: 

41 return ImpactResult( 

42 node_id=node_id, 

43 impacted_nodes=[], 

44 impact_depth=0, 

45 impact_score=0.0, 

46 ) 

47 

48 # Get all downstream nodes 

49 downstream = self.engine.get_downstream_nodes(node_id, max_depth) 

50 

51 # Calculate impact score based on criticality 

52 impact_score = self._calculate_impact_score(downstream) 

53 

54 # Group by severity and type 

55 by_severity = self._group_by_severity(downstream) 

56 by_type = self._group_by_type(downstream) 

57 

58 # Calculate depth 

59 depth = self._calculate_max_depth(node_id, downstream) 

60 

61 return ImpactResult( 

62 node_id=node_id, 

63 impacted_nodes=downstream, 

64 impact_depth=depth, 

65 impact_score=impact_score, 

66 by_severity=by_severity, 

67 by_type=by_type, 

68 metadata={ 

69 "total_impacted": len(downstream), 

70 "analyzed_at": datetime.utcnow().isoformat(), 

71 }, 

72 ) 

73 

74 def _calculate_impact_score(self, nodes: List[str]) -> float: 

75 """Calculate impact score 0.0-1.0 based on criticality.""" 

76 if not nodes: 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true

77 return 0.0 

78 

79 critical_count = sum( 

80 1 

81 for n in nodes 

82 if self.graph.nodes.get(n, Node("", "", NodeType.TASK)).severity 

83 == NodeSeverity.CRITICAL 

84 ) 

85 

86 return min(1.0, critical_count / len(nodes)) 

87 

88 def _group_by_severity(self, nodes: List[str]) -> Dict[NodeSeverity, List[str]]: 

89 """Group nodes by severity.""" 

90 result = {} 

91 for severity in NodeSeverity: 

92 result[severity] = [ 

93 n 

94 for n in nodes 

95 if self.graph.nodes.get(n, Node("", "", NodeType.TASK)).severity == severity 

96 ] 

97 return {k: v for k, v in result.items() if v} 

98 

99 def _group_by_type(self, nodes: List[str]) -> Dict[NodeType, List[str]]: 

100 """Group nodes by type.""" 

101 result = {} 

102 for node_type in NodeType: 

103 result[node_type] = [ 

104 n 

105 for n in nodes 

106 if self.graph.nodes.get(n, Node("", "", NodeType.TASK)).type == node_type 

107 ] 

108 return {k: v for k, v in result.items() if v} 

109 

110 def _calculate_max_depth(self, source: str, nodes: List[str]) -> int: 

111 """Calculate maximum depth to any impacted node.""" 

112 if not nodes: 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true

113 return 0 

114 

115 max_d = 0 

116 for node_id in nodes: 

117 path = self.engine.get_path(source, node_id) 

118 if path: 118 ↛ 116line 118 didn't jump to line 116 because the condition on line 118 was always true

119 max_d = max(max_d, len(path) - 1) 

120 

121 return max_d 

122 

123 

124class BlastRadiusEngine: 

125 """Analyze blast radius of changes.""" 

126 

127 def __init__(self, graph: DependencyGraph): 

128 self.graph = graph 

129 self.engine = DependencyGraphEngine(graph) 

130 

131 def analyze(self, change_nodes: List[str]) -> BlastRadiusResult: 

132 """ 

133 Analyze blast radius of multiple node changes. 

134 

135 Args: 

136 change_nodes: List of nodes that changed 

137 

138 Returns: 

139 BlastRadiusResult with affected nodes and risk assessment 

140 """ 

141 # Get all affected nodes (union of downstream) 

142 affected = set() 

143 for node_id in change_nodes: 

144 affected.update(self.engine.get_downstream_nodes(node_id)) 

145 

146 # Calculate severity distribution 

147 severity_dist = self._calculate_severity_distribution(list(affected)) 

148 

149 # Determine risk level 

150 risk_level = self._assess_risk_level(list(affected), severity_dist) 

151 

152 # Check if deployment is safe 

153 deployable = self._is_deployable(list(affected), risk_level) 

154 

155 # Calculate blast depth 

156 blast_depth = self._calculate_blast_depth(change_nodes, list(affected)) 

157 

158 return BlastRadiusResult( 

159 change_nodes=change_nodes, 

160 affected_nodes=list(affected), 

161 blast_radius=len(affected), 

162 blast_depth=blast_depth, 

163 severity_distribution=severity_dist, 

164 risk_level=risk_level, 

165 deployable=deployable, 

166 metadata={ 

167 "analyzed_at": datetime.utcnow().isoformat(), 

168 "change_count": len(change_nodes), 

169 }, 

170 ) 

171 

172 def _calculate_severity_distribution(self, nodes: List[str]) -> Dict[NodeSeverity, int]: 

173 """Calculate distribution of node severities.""" 

174 dist = {s: 0 for s in NodeSeverity} 

175 for node_id in nodes: 

176 node = self.graph.nodes.get(node_id) 

177 if node: 177 ↛ 175line 177 didn't jump to line 175 because the condition on line 177 was always true

178 dist[node.severity] += 1 

179 return {k: v for k, v in dist.items() if v > 0} 

180 

181 def _assess_risk_level(self, nodes: List[str], severity_dist: Dict[NodeSeverity, int]) -> str: 

182 """Assess risk level based on affected nodes.""" 

183 if severity_dist.get(NodeSeverity.CRITICAL, 0) > 0: 183 ↛ 184line 183 didn't jump to line 184 because the condition on line 183 was never true

184 return "critical" 

185 elif severity_dist.get(NodeSeverity.HIGH, 0) > 3: 185 ↛ 186line 185 didn't jump to line 186 because the condition on line 185 was never true

186 return "high" 

187 elif severity_dist.get(NodeSeverity.MEDIUM, 0) > 10: 187 ↛ 188line 187 didn't jump to line 188 because the condition on line 187 was never true

188 return "medium" 

189 else: 

190 return "low" 

191 

192 def _is_deployable(self, nodes: List[str], risk_level: str) -> bool: 

193 """Determine if changes are safe to deploy.""" 

194 return risk_level not in ("critical", "high") 

195 

196 def _calculate_blast_depth(self, change_nodes: List[str], affected: List[str]) -> int: 

197 """Calculate maximum depth of blast.""" 

198 if not affected: 

199 return 0 

200 

201 max_depth = 0 

202 for change_node in change_nodes: 

203 for affected_node in affected: 

204 path = self.engine.get_path(change_node, affected_node) 

205 if path: 205 ↛ 203line 205 didn't jump to line 203 because the condition on line 205 was always true

206 max_depth = max(max_depth, len(path) - 1) 

207 

208 return max_depth 

209 

210 

211class RiskScoringEngine: 

212 """Calculate risk scores for nodes based on criticality and dependencies.""" 

213 

214 def __init__(self, graph: DependencyGraph): 

215 self.graph = graph 

216 self.engine = DependencyGraphEngine(graph) 

217 

218 def score_node(self, node_id: str) -> RiskScoreResult: 

219 """ 

220 Calculate risk score for a node. 

221 

222 Risk factors: 

223 - Severity (0-2 points) 

224 - Downstream count (0-3 points) 

225 - Upstream count (0-2 points) 

226 - Critical dependents (0-3 points) 

227 

228 Returns: 

229 RiskScoreResult with score and breakdown 

230 """ 

231 node = self.graph.nodes.get(node_id) 

232 if not node: 232 ↛ 233line 232 didn't jump to line 233 because the condition on line 232 was never true

233 return RiskScoreResult( 

234 node_id=node_id, 

235 risk_score=0.0, 

236 components={}, 

237 factors=[], 

238 severity=NodeSeverity.LOW, 

239 ) 

240 

241 components = {} 

242 factors = [] 

243 

244 # Severity score (0-2) 

245 severity_scores = { 

246 NodeSeverity.CRITICAL: 2.0, 

247 NodeSeverity.HIGH: 1.5, 

248 NodeSeverity.MEDIUM: 1.0, 

249 NodeSeverity.LOW: 0.5, 

250 } 

251 components["severity"] = severity_scores.get(node.severity, 0.0) 

252 if node.severity in (NodeSeverity.CRITICAL, NodeSeverity.HIGH): 

253 factors.append(f"High severity: {node.severity.value}") 

254 

255 # Downstream impact (0-3) 

256 downstream = self.engine.get_downstream_nodes(node_id) 

257 downstream_score = min(3.0, len(downstream) / 10.0) 

258 components["downstream"] = downstream_score 

259 if len(downstream) > 50: 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true

260 factors.append(f"High downstream impact: {len(downstream)} nodes") 

261 

262 # Upstream criticality (0-2) 

263 upstream = self.engine.get_upstream_nodes(node_id) 

264 critical_upstream = sum( 

265 1 

266 for u in upstream 

267 if self.graph.nodes.get(u, Node("", "", NodeType.TASK)).severity 

268 == NodeSeverity.CRITICAL 

269 ) 

270 upstream_score = min(2.0, critical_upstream / 5.0) 

271 components["upstream"] = upstream_score 

272 if critical_upstream > 0: 

273 factors.append(f"Depends on {critical_upstream} critical upstream nodes") 

274 

275 # Critical dependents (0-3) 

276 critical_downstream = sum( 

277 1 

278 for d in downstream 

279 if self.graph.nodes.get(d, Node("", "", NodeType.TASK)).severity 

280 == NodeSeverity.CRITICAL 

281 ) 

282 critical_score = min(3.0, critical_downstream / 5.0) 

283 components["critical_dependents"] = critical_score 

284 if critical_downstream > 0: 284 ↛ 285line 284 didn't jump to line 285 because the condition on line 284 was never true

285 factors.append(f"Has {critical_downstream} critical downstream nodes") 

286 

287 # Cycle involvement 

288 cycles = self.engine.detect_cycles() 

289 in_cycle = any(node_id in cycle for cycle in cycles) 

290 cycle_score = 2.0 if in_cycle else 0.0 

291 components["in_cycle"] = cycle_score 

292 if in_cycle: 292 ↛ 293line 292 didn't jump to line 293 because the condition on line 292 was never true

293 factors.append("Node is part of a circular dependency") 

294 

295 # Calculate total score (0-10) 

296 total_score = sum(components.values()) 

297 risk_score = min(10.0, total_score) 

298 

299 return RiskScoreResult( 

300 node_id=node_id, 

301 risk_score=risk_score, 

302 components=components, 

303 factors=factors, 

304 severity=self._score_to_severity(risk_score), 

305 metadata={ 

306 "calculated_at": datetime.utcnow().isoformat(), 

307 "upstream_nodes": len(upstream), 

308 "downstream_nodes": len(downstream), 

309 }, 

310 ) 

311 

312 def _score_to_severity(self, score: float) -> NodeSeverity: 

313 """Convert risk score to severity level.""" 

314 if score >= 8.0: 314 ↛ 315line 314 didn't jump to line 315 because the condition on line 314 was never true

315 return NodeSeverity.CRITICAL 

316 elif score >= 6.0: 316 ↛ 317line 316 didn't jump to line 317 because the condition on line 316 was never true

317 return NodeSeverity.HIGH 

318 elif score >= 3.0: 318 ↛ 319line 318 didn't jump to line 319 because the condition on line 318 was never true

319 return NodeSeverity.MEDIUM 

320 else: 

321 return NodeSeverity.LOW 

322 

323 def score_all_nodes(self) -> Dict[str, RiskScoreResult]: 

324 """Calculate risk scores for all nodes.""" 

325 return {node_id: self.score_node(node_id) for node_id in self.graph.nodes} 

326 

327 

328class DriftDetectionEngine: 

329 """Detect changes in dependencies (drift detection).""" 

330 

331 def __init__( 

332 self, 

333 current_graph: DependencyGraph, 

334 previous_graph: Optional[DependencyGraph] = None, 

335 ): 

336 self.current_graph = current_graph 

337 self.previous_graph = previous_graph or DependencyGraph() 

338 

339 def detect_drift(self) -> DriftDetectionResult: 

340 """ 

341 Detect dependency drift between current and previous graphs. 

342 

343 Returns: 

344 DriftDetectionResult with detected changes 

345 """ 

346 drifts = [] 

347 affected_nodes = set() 

348 details = [] 

349 

350 # Detect added nodes 

351 added_nodes = set(self.current_graph.nodes.keys()) - set(self.previous_graph.nodes.keys()) 

352 if added_nodes: 

353 drifts.append( 

354 { 

355 "type": "nodes_added", 

356 "count": len(added_nodes), 

357 "nodes": list(added_nodes), 

358 } 

359 ) 

360 affected_nodes.update(added_nodes) 

361 details.append(f"Added {len(added_nodes)} new nodes") 

362 

363 # Detect removed nodes 

364 removed_nodes = set(self.previous_graph.nodes.keys()) - set(self.current_graph.nodes.keys()) 

365 if removed_nodes: 

366 drifts.append( 

367 { 

368 "type": "nodes_removed", 

369 "count": len(removed_nodes), 

370 "nodes": list(removed_nodes), 

371 } 

372 ) 

373 affected_nodes.update(removed_nodes) 

374 details.append(f"Removed {len(removed_nodes)} nodes") 

375 

376 # Detect added edges 

377 prev_edges = {(e.source, e.target) for e in self.previous_graph.edges} 

378 curr_edges = {(e.source, e.target) for e in self.current_graph.edges} 

379 added_edges = curr_edges - prev_edges 

380 

381 if added_edges: 

382 drifts.append( 

383 { 

384 "type": "edges_added", 

385 "count": len(added_edges), 

386 "edges": [{"source": s, "target": t} for s, t in sorted(added_edges)], 

387 } 

388 ) 

389 for source, target in added_edges: 

390 affected_nodes.add(source) 

391 affected_nodes.add(target) 

392 details.append(f"Added {len(added_edges)} new dependencies") 

393 

394 # Detect removed edges 

395 removed_edges = prev_edges - curr_edges 

396 if removed_edges: 

397 drifts.append( 

398 { 

399 "type": "edges_removed", 

400 "count": len(removed_edges), 

401 "edges": [{"source": s, "target": t} for s, t in sorted(removed_edges)], 

402 } 

403 ) 

404 for source, target in removed_edges: 

405 affected_nodes.add(source) 

406 affected_nodes.add(target) 

407 details.append(f"Removed {len(removed_edges)} dependencies") 

408 

409 # Determine severity based on drift 

410 drift_count = len(drifts) 

411 if drift_count == 0: 

412 severity = NodeSeverity.LOW 

413 elif added_edges or removed_edges: 

414 # Edge changes are more significant than node changes 

415 severity = NodeSeverity.HIGH 

416 elif removed_nodes: 416 ↛ 418line 416 didn't jump to line 418 because the condition on line 416 was never true

417 # Removed nodes are concerning 

418 severity = NodeSeverity.MEDIUM 

419 else: 

420 severity = NodeSeverity.LOW 

421 

422 return DriftDetectionResult( 

423 detected_drifts=drifts, 

424 drift_count=drift_count, 

425 affected_nodes=list(affected_nodes), 

426 severity=severity, 

427 details=details, 

428 metadata={ 

429 "detected_at": datetime.utcnow().isoformat(), 

430 "previous_nodes": len(self.previous_graph.nodes), 

431 "current_nodes": len(self.current_graph.nodes), 

432 "previous_edges": len(self.previous_graph.edges), 

433 "current_edges": len(self.current_graph.edges), 

434 }, 

435 )