Batch-fetch nodes by qualified name, staying under SQLite variable limits.
(self, qualified_names: set[str])
| 1227 | return results |
| 1228 | |
| 1229 | def _batch_get_nodes(self, qualified_names: set[str]) -> list[GraphNode]: |
| 1230 | """Batch-fetch nodes by qualified name, staying under SQLite variable limits.""" |
| 1231 | if not qualified_names: |
| 1232 | return [] |
| 1233 | qns = list(qualified_names) |
| 1234 | results: list[GraphNode] = [] |
| 1235 | batch_size = 450 |
| 1236 | for i in range(0, len(qns), batch_size): |
| 1237 | batch = qns[i:i + batch_size] |
| 1238 | placeholders = ",".join("?" for _ in batch) |
| 1239 | rows = self._conn.execute( # nosec B608 |
| 1240 | f"SELECT * FROM nodes WHERE qualified_name IN ({placeholders})", |
| 1241 | batch, |
| 1242 | ).fetchall() |
| 1243 | results.extend(self._row_to_node(r) for r in rows) |
| 1244 | return results |
| 1245 | |
| 1246 | def load_flow_adjacency(self) -> "FlowAdjacency": |
| 1247 | """Load all nodes and CALLS/TESTED_BY edges into memory for fast traversal. |
no test coverage detected