Coverage for python/pyairflowtester/dependency_intelligence/graph.py: 95%

1"""Core dependency graph engine with traversal and analysis algorithms.""" 

2 

3import hashlib 

4import logging 

5from collections import deque 

6from typing import Any, Dict, List, Optional, Set 

7 

8from .cache import TieredCache 

9from .models import ( 

10 DependencyGraph, 

11 NodeType, 

12) 

13 

14logger = logging.getLogger(__name__) 

15 

16 

17class DependencyGraphEngine: 

18 """Core graph engine with algorithms for traversal and analysis.""" 

19 

20 def __init__(self, graph: DependencyGraph, cache: Optional[TieredCache] = None): 

21 """ 

22 Args: 

23 graph: The dependency graph to analyze. 

24 cache: Optional TieredCache (see cache.py) for persisting 

25 expensive analyses (cycle detection, strongly connected 

26 components) across calls -- and, with an L3 SqliteCache, 

27 across separate process invocations, which the in-process 

28 `_cycles_cache` below can never do. Cache keys are content 

29 hashes of the graph's nodes/edges, so a changed graph 

30 naturally misses rather than serving stale results. 

31 """ 

32 self.graph = graph 

33 self.cache = cache 

34 self._upstream_cache: Dict[str, Set[str]] = {} 

35 self._downstream_cache: Dict[str, Set[str]] = {} 

36 self._cycles_cache: Optional[List[List[str]]] = None 

37 self._cache_valid = True 

38 

39 def _graph_content_hash(self) -> str: 

40 """Stable hash of the graph's current nodes/edges, used as a cache 

41 key prefix so a modified graph never serves another graph's -- or 

42 its own earlier, since-changed -- cached analysis results.""" 

43 node_ids = sorted(self.graph.nodes.keys()) 

44 edge_pairs = sorted((e.source, e.target) for e in self.graph.edges) 

45 digest_input = "|".join(node_ids) + "::" + "|".join(f"{s}>{t}" for s, t in edge_pairs) 

46 return hashlib.sha256(digest_input.encode("utf-8")).hexdigest()[:16] 

47 

48 def invalidate_cache(self): 

49 """Invalidate all caches after graph modification.""" 

50 self._upstream_cache.clear() 

51 self._downstream_cache.clear() 

52 self._cycles_cache = None 

53 self._cache_valid = False 

54 

55 # Traversal algorithms 

56 

57 def get_upstream_nodes(self, node_id: str, max_depth: Optional[int] = None) -> List[str]: 

58 """ 

59 Get all upstream nodes (dependencies) of a node using BFS. 

60 

61 Args: 

62 node_id: The node to analyze 

63 max_depth: Optional maximum traversal depth 

64 

65 Returns: 

66 List of upstream node IDs 

67 """ 

68 if node_id not in self.graph.nodes: 

69 return [] 

70 

71 # Check cache 

72 if node_id in self._upstream_cache and max_depth is None: 

73 return list(self._upstream_cache[node_id]) 

74 

75 visited = set() 

76 queue = deque([(node_id, 0)]) 

77 upstream = set() 

78 

79 while queue: 

80 current_id, depth = queue.popleft() 

81 

82 # Check depth limit 

83 if max_depth is not None and depth >= max_depth: 

84 continue 

85 

86 # Get incoming edges (dependencies) 

87 for edge in self.graph.get_edges_to(current_id): 

88 source_id = edge.source 

89 

90 if source_id not in visited: 90 ↛ 87line 90 didn't jump to line 87 because the condition on line 90 was always true

91 visited.add(source_id) 

92 upstream.add(source_id) 

93 queue.append((source_id, depth + 1)) 

94 

95 # Cache if no depth limit 

96 if max_depth is None: 

97 self._upstream_cache[node_id] = upstream 

98 

99 return sorted(list(upstream)) 

100 

101 def get_downstream_nodes(self, node_id: str, max_depth: Optional[int] = None) -> List[str]: 

102 """ 

103 Get all downstream nodes (dependents) of a node using BFS. 

104 

105 Args: 

106 node_id: The node to analyze 

107 max_depth: Optional maximum traversal depth 

108 

109 Returns: 

110 List of downstream node IDs 

111 """ 

112 if node_id not in self.graph.nodes: 

113 return [] 

114 

115 # Check cache 

116 if node_id in self._downstream_cache and max_depth is None: 

117 return list(self._downstream_cache[node_id]) 

118 

119 visited = set() 

120 queue = deque([(node_id, 0)]) 

121 downstream = set() 

122 

123 while queue: 

124 current_id, depth = queue.popleft() 

125 

126 # Check depth limit 

127 if max_depth is not None and depth >= max_depth: 

128 continue 

129 

130 # Get outgoing edges (dependents) 

131 for edge in self.graph.get_edges_from(current_id): 

132 target_id = edge.target 

133 

134 if target_id not in visited: 134 ↛ 131line 134 didn't jump to line 131 because the condition on line 134 was always true

135 visited.add(target_id) 

136 downstream.add(target_id) 

137 queue.append((target_id, depth + 1)) 

138 

139 # Cache if no depth limit 

140 if max_depth is None: 

141 self._downstream_cache[node_id] = downstream 

142 

143 return sorted(list(downstream)) 

144 

145 def get_reach(self, node_id: str) -> Dict[str, int]: 

146 """ 

147 Get reachability from a node with distances. 

148 

149 Returns: 

150 Dict of {node_id: distance} 

151 """ 

152 if node_id not in self.graph.nodes: 152 ↛ 153line 152 didn't jump to line 153 because the condition on line 152 was never true

153 return {} 

154 

155 visited = {node_id: 0} 

156 queue = deque([(node_id, 0)]) 

157 

158 while queue: 

159 current_id, distance = queue.popleft() 

160 

161 for edge in self.graph.get_edges_from(current_id): 

162 target_id = edge.target 

163 

164 if target_id not in visited: 164 ↛ 161line 164 didn't jump to line 161 because the condition on line 164 was always true

165 visited[target_id] = distance + 1 

166 queue.append((target_id, distance + 1)) 

167 

168 return visited 

169 

170 def get_path(self, source_id: str, target_id: str) -> Optional[List[str]]: 

171 """ 

172 Find shortest path between two nodes using BFS. 

173 

174 Returns: 

175 List of node IDs from source to target, or None if no path 

176 """ 

177 if source_id not in self.graph.nodes or target_id not in self.graph.nodes: 177 ↛ 178line 177 didn't jump to line 178 because the condition on line 177 was never true

178 return None 

179 

180 if source_id == target_id: 

181 return [source_id] 

182 

183 visited = {source_id} 

184 queue = deque([(source_id, [source_id])]) 

185 

186 while queue: 

187 current_id, path = queue.popleft() 

188 

189 for edge in self.graph.get_edges_from(current_id): 

190 neighbor_id = edge.target 

191 

192 if neighbor_id == target_id: 

193 return path + [neighbor_id] 

194 

195 if neighbor_id not in visited: 195 ↛ 189line 195 didn't jump to line 189 because the condition on line 195 was always true

196 visited.add(neighbor_id) 

197 queue.append((neighbor_id, path + [neighbor_id])) 

198 

199 return None 

200 

201 # Cycle detection 

202 

203 def detect_cycles(self) -> List[List[str]]: 

204 """ 

205 Detect all cycles in the dependency graph using DFS. 

206 

207 Returns: 

208 List of cycles, where each cycle is a list of node IDs 

209 """ 

210 if self._cycles_cache is not None: 

211 return self._cycles_cache 

212 

213 cache_key = None 

214 if self.cache is not None: 

215 cache_key = f"cycles:{self._graph_content_hash()}" 

216 cached = self.cache.get(cache_key) 

217 if cached is not None: 

218 self._cycles_cache = cached 

219 return cached 

220 

221 cycles = [] 

222 visited = set() 

223 rec_stack = set() 

224 path_stack = [] 

225 

226 def dfs(node_id: str): 

227 visited.add(node_id) 

228 rec_stack.add(node_id) 

229 path_stack.append(node_id) 

230 

231 for edge in self.graph.get_edges_from(node_id): 

232 neighbor_id = edge.target 

233 

234 if neighbor_id not in visited: 

235 dfs(neighbor_id) 

236 elif neighbor_id in rec_stack: 236 ↛ 231line 236 didn't jump to line 231 because the condition on line 236 was always true

237 # Found a cycle 

238 cycle_start_idx = path_stack.index(neighbor_id) 

239 cycle = path_stack[cycle_start_idx:] + [neighbor_id] 

240 cycles.append(cycle) 

241 

242 path_stack.pop() 

243 rec_stack.discard(node_id) 

244 

245 # Run DFS from all unvisited nodes 

246 for node_id in self.graph.nodes: 

247 if node_id not in visited: 

248 dfs(node_id) 

249 

250 # Cache result 

251 self._cycles_cache = cycles 

252 if self.cache is not None and cache_key is not None: 

253 self.cache.set(cache_key, cycles, ttl_seconds=600) 

254 return cycles 

255 

256 def has_cycle(self) -> bool: 

257 """Check if graph has any cycles.""" 

258 return len(self.detect_cycles()) > 0 

259 

260 # Orphan detection 

261 

262 def detect_orphans(self) -> Dict[str, List[str]]: 

263 """ 

264 Detect orphaned nodes (no incoming/outgoing edges). 

265 

266 Returns: 

267 Dict with 'sources' (no incoming) and 'sinks' (no outgoing) 

268 """ 

269 sources = [] # No incoming edges 

270 sinks = [] # No outgoing edges 

271 

272 for node_id, node in self.graph.nodes.items(): 

273 has_incoming = any(e.target == node_id for e in self.graph.edges) 

274 has_outgoing = any(e.source == node_id for e in self.graph.edges) 

275 

276 if not has_incoming: 

277 sources.append(node_id) 

278 if not has_outgoing: 

279 sinks.append(node_id) 

280 

281 return { 

282 "sources": sorted(sources), 

283 "sinks": sorted(sinks), 

284 "isolated": sorted([n for n in sources if n in sinks]), 

285 } 

286 

287 def detect_disconnected_components(self) -> List[Set[str]]: 

288 """ 

289 Find disconnected components in the graph. 

290 

291 Returns: 

292 List of sets, each containing node IDs in one component 

293 """ 

294 visited = set() 

295 components = [] 

296 

297 def dfs(node_id: str, component: Set[str]): 

298 visited.add(node_id) 

299 component.add(node_id) 

300 

301 # Check both incoming and outgoing edges 

302 for edge in self.graph.edges: 

303 if edge.source == node_id and edge.target not in visited: 

304 dfs(edge.target, component) 

305 elif edge.target == node_id and edge.source not in visited: 305 ↛ 306line 305 didn't jump to line 306 because the condition on line 305 was never true

306 dfs(edge.source, component) 

307 

308 for node_id in self.graph.nodes: 

309 if node_id not in visited: 

310 component = set() 

311 dfs(node_id, component) 

312 components.append(component) 

313 

314 return components 

315 

316 # Advanced analysis 

317 

318 def get_critical_path(self) -> List[str]: 

319 """ 

320 Find the longest path in the DAG (critical path). 

321 For cyclic graphs, returns longest acyclic path. 

322 """ 

323 if self.has_cycle(): 323 ↛ 324line 323 didn't jump to line 324 because the condition on line 323 was never true

324 logger.warning("Graph has cycles; returning longest acyclic path") 

325 

326 max_path = [] 

327 visited = set() 

328 

329 def dfs(node_id: str, path: List[str]) -> List[str]: 

330 nonlocal max_path 

331 visited.add(node_id) 

332 

333 if len(path) > len(max_path): 

334 max_path = path[:] 

335 

336 for edge in self.graph.get_edges_from(node_id): 

337 if edge.target not in visited: 337 ↛ 336line 337 didn't jump to line 336 because the condition on line 337 was always true

338 path.append(edge.target) 

339 dfs(edge.target, path) 

340 path.pop() 

341 

342 visited.discard(node_id) 

343 

344 for node_id in self.graph.nodes: 

345 path = [node_id] 

346 dfs(node_id, path) 

347 

348 return max_path 

349 

350 def get_strongly_connected_components(self) -> List[Set[str]]: 

351 """ 

352 Find strongly connected components using Tarjan's algorithm. 

353 For dependency graphs, identifies circular dependencies. 

354 """ 

355 cache_key = None 

356 if self.cache is not None: 356 ↛ 362line 356 didn't jump to line 362 because the condition on line 356 was always true

357 cache_key = f"sccs:{self._graph_content_hash()}" 

358 cached = self.cache.get(cache_key) 

359 if cached is not None: 

360 return [set(scc) for scc in cached] 

361 

362 index = 0 

363 stack = [] 

364 indices = {} 

365 lowlinks = {} 

366 on_stack = set() 

367 sccs = [] 

368 

369 def strongconnect(node_id: str): 

370 nonlocal index 

371 indices[node_id] = index 

372 lowlinks[node_id] = index 

373 index += 1 

374 stack.append(node_id) 

375 on_stack.add(node_id) 

376 

377 for edge in self.graph.get_edges_from(node_id): 

378 target_id = edge.target 

379 if target_id not in indices: 

380 strongconnect(target_id) 

381 lowlinks[node_id] = min(lowlinks[node_id], lowlinks[target_id]) 

382 elif target_id in on_stack: 382 ↛ 377line 382 didn't jump to line 377 because the condition on line 382 was always true

383 lowlinks[node_id] = min(lowlinks[node_id], indices[target_id]) 

384 

385 if lowlinks[node_id] == indices[node_id]: 

386 scc = set() 

387 while True: 

388 node = stack.pop() 

389 on_stack.discard(node) 

390 scc.add(node) 

391 if node == node_id: 

392 break 

393 sccs.append(scc) 

394 

395 for node_id in self.graph.nodes: 

396 if node_id not in indices: 

397 strongconnect(node_id) 

398 

399 if self.cache is not None and cache_key is not None: 399 ↛ 401line 399 didn't jump to line 401 because the condition on line 399 was always true

400 self.cache.set(cache_key, [sorted(scc) for scc in sccs], ttl_seconds=600) 

401 return sccs 

402 

403 def get_node_centrality(self) -> Dict[str, float]: 

404 """ 

405 Calculate degree centrality for each node. 

406 

407 Returns: 

408 Dict of {node_id: centrality_score} 

409 """ 

410 max_degree = len(self.graph.nodes) - 1 

411 if max_degree == 0: 411 ↛ 412line 411 didn't jump to line 412 because the condition on line 411 was never true

412 return {n: 0.0 for n in self.graph.nodes} 

413 

414 centrality = {} 

415 for node_id in self.graph.nodes: 

416 upstream = len(self.get_upstream_nodes(node_id)) 

417 downstream = len(self.get_downstream_nodes(node_id)) 

418 total_degree = upstream + downstream 

419 centrality[node_id] = total_degree / (2 * max_degree) 

420 

421 return centrality 

422 

423 def filter_by_type(self, node_type: NodeType) -> List[str]: 

424 """Get all nodes of a specific type.""" 

425 return [n.id for n in self.graph.nodes.values() if n.type == node_type] 

426 

427 def filter_by_owner(self, owner: str) -> List[str]: 

428 """Get all nodes owned by a specific owner.""" 

429 return [n.id for n in self.graph.nodes.values() if n.owner == owner] 

430 

431 def get_stats(self) -> Dict[str, Any]: 

432 """Get comprehensive graph statistics.""" 

433 cycles = self.detect_cycles() 

434 orphans = self.detect_orphans() 

435 components = self.detect_disconnected_components() 

436 centrality = self.get_node_centrality() 

437 

438 return { 

439 "node_count": len(self.graph.nodes), 

440 "edge_count": len(self.graph.edges), 

441 "cycle_count": len(cycles), 

442 "has_cycles": len(cycles) > 0, 

443 "component_count": len(components), 

444 "is_connected": len(components) <= 1, 

445 "orphaned_sources": len(orphans["sources"]), 

446 "orphaned_sinks": len(orphans["sinks"]), 

447 "orphaned_isolated": len(orphans["isolated"]), 

448 "average_degree": ( 

449 sum( 

450 len(self.get_upstream_nodes(n)) + len(self.get_downstream_nodes(n)) 

451 for n in self.graph.nodes 

452 ) 

453 / len(self.graph.nodes) 

454 if self.graph.nodes 

455 else 0 

456 ), 

457 "avg_centrality": sum(centrality.values()) / len(centrality) if centrality else 0, 

458 "max_centrality": max(centrality.values()) if centrality else 0, 

459 "node_types": self.graph.get_node_count_by_type(), 

460 }