In-memory representation of the Code Graph for architectural simulations.
| 21 | |
| 22 | |
| 23 | class CodeGraphTwin: |
| 24 | """In-memory representation of the Code Graph for architectural simulations.""" |
| 25 | |
| 26 | def __init__(self, repository_path: str): |
| 27 | self.repository_path = Path(repository_path).resolve().as_posix() |
| 28 | self.nodes: Dict[str, Dict[str, Any]] = {} |
| 29 | self.edges: List[Dict[str, Any]] = [] |
| 30 | self.service_mapping: Dict[str, str] = {} |
| 31 | self.repo_name = Path(repository_path).name |
| 32 | |
| 33 | def load_from_db(self, db_manager) -> "CodeGraphTwin": |
| 34 | """Fetches the repository structure and relationships from the database.""" |
| 35 | info_logger(f"Loading digital twin from DB for repo path: {self.repository_path}") |
| 36 | |
| 37 | # 1. Fetch all nodes contained by the repository |
| 38 | node_query = """ |
| 39 | MATCH (r:Repository {path: $path}) |
| 40 | OPTIONAL MATCH (r)-[:CONTAINS*]->(n) |
| 41 | RETURN |
| 42 | r.path as repo_path, r.name as repo_name, |
| 43 | labels(n)[0] as label, |
| 44 | n.uid as uid, |
| 45 | n.name as name, |
| 46 | n.path as path, |
| 47 | n.cyclomatic_complexity as complexity |
| 48 | """ |
| 49 | |
| 50 | # 2. Fetch all relationships originating from nodes in the repository |
| 51 | rel_query = """ |
| 52 | MATCH (r:Repository {path: $path}) |
| 53 | MATCH (n1) WHERE n1 = r OR (r)-[:CONTAINS*]->(n1) |
| 54 | MATCH (n1)-[rel]->(n2) |
| 55 | RETURN |
| 56 | labels(n1)[0] as source_label, |
| 57 | n1.uid as source_uid, |
| 58 | n1.path as source_path, |
| 59 | n1.name as source_name, |
| 60 | type(rel) as rel_type, |
| 61 | labels(n2)[0] as target_label, |
| 62 | n2.uid as target_uid, |
| 63 | n2.path as target_path, |
| 64 | n2.name as target_name |
| 65 | """ |
| 66 | |
| 67 | with db_manager.get_driver().session() as session: |
| 68 | node_rows = session.run(node_query, path=self.repository_path).data() |
| 69 | rel_rows = session.run(rel_query, path=self.repository_path).data() |
| 70 | |
| 71 | # Build nodes |
| 72 | for row in node_rows: |
| 73 | label = row.get("label") |
| 74 | if not label: |
| 75 | # If repository is empty or n is null |
| 76 | if row.get("repo_name"): |
| 77 | self.repo_name = row.get("repo_name") |
| 78 | continue |
| 79 | |
| 80 | node_id = resolve_node_id(row) |
no outgoing calls