Handles creation and loading of .cgc bundle files.
| 51 | |
| 52 | |
| 53 | class CGCBundle: |
| 54 | """Handles creation and loading of .cgc bundle files.""" |
| 55 | |
| 56 | VERSION = "0.1.0" # CGC bundle format version |
| 57 | |
| 58 | def __init__(self, db_manager): |
| 59 | """ |
| 60 | Initialize the CGC Bundle handler. |
| 61 | |
| 62 | Args: |
| 63 | db_manager: DatabaseManager instance for graph queries |
| 64 | """ |
| 65 | self.db_manager = db_manager |
| 66 | |
| 67 | def _get_id_function(self) -> str: |
| 68 | """ |
| 69 | Get the appropriate ID function based on the database backend. |
| 70 | |
| 71 | Returns: |
| 72 | str: 'elementId' for Neo4j, 'id' for FalkorDB |
| 73 | """ |
| 74 | backend = self.db_manager.get_backend_type() |
| 75 | if backend == 'neo4j': |
| 76 | return 'elementId' |
| 77 | return 'id' |
| 78 | |
| 79 | def _uses_pk_edge_matching(self) -> bool: |
| 80 | """Kùzu/Ladybug internal IDs are not comparable via id() in MATCH.""" |
| 81 | return self.db_manager.get_backend_type() in {'kuzudb', 'ladybugdb'} |
| 82 | |
| 83 | def _node_lookup_key(self, labels, properties: Dict) -> Optional[tuple]: |
| 84 | if not labels: |
| 85 | return None |
| 86 | if isinstance(labels, str): |
| 87 | labels = [labels] |
| 88 | primary_label = labels[0] |
| 89 | pk_field = self._PK_MAP.get(primary_label) |
| 90 | if pk_field and pk_field in properties: |
| 91 | return (primary_label, pk_field, properties[pk_field]) |
| 92 | return None |
| 93 | |
| 94 | |
| 95 | def export_to_bundle( |
| 96 | self, |
| 97 | output_path: Path, |
| 98 | repo_path: Optional[Path] = None, |
| 99 | include_stats: bool = True |
| 100 | ) -> Tuple[bool, str]: |
| 101 | """ |
| 102 | Export the current graph (or a specific repository) to a .cgc bundle. |
| 103 | |
| 104 | Args: |
| 105 | output_path: Path where the .cgc file should be saved |
| 106 | repo_path: Optional specific repository path to export (None = export all) |
| 107 | include_stats: Whether to include detailed statistics |
| 108 | |
| 109 | Returns: |
| 110 | Tuple[bool, str]: (success, message) |
no outgoing calls
no test coverage detected