Handles creation and loading of .cgc bundle files.
| 50 | |
| 51 | |
| 52 | class CGCBundle: |
| 53 | """Handles creation and loading of .cgc bundle files.""" |
| 54 | |
| 55 | VERSION = "0.1.0" # CGC bundle format version |
| 56 | |
| 57 | def __init__(self, db_manager): |
| 58 | """ |
| 59 | Initialize the CGC Bundle handler. |
| 60 | |
| 61 | Args: |
| 62 | db_manager: DatabaseManager instance for graph queries |
| 63 | """ |
| 64 | self.db_manager = db_manager |
| 65 | |
| 66 | def _get_id_function(self) -> str: |
| 67 | """ |
| 68 | Get the appropriate ID function based on the database backend. |
| 69 | |
| 70 | Returns: |
| 71 | str: 'elementId' for Neo4j, 'id' for FalkorDB |
| 72 | """ |
| 73 | # Check if we're using Neo4j or FalkorDB |
| 74 | backend = self.db_manager.get_backend_type() |
| 75 | if backend == 'neo4j': |
| 76 | return 'elementId' |
| 77 | else: # FalkorDB or other backends |
| 78 | return 'id' |
| 79 | |
| 80 | |
| 81 | def export_to_bundle( |
| 82 | self, |
| 83 | output_path: Path, |
| 84 | repo_path: Optional[Path] = None, |
| 85 | include_stats: bool = True |
| 86 | ) -> Tuple[bool, str]: |
| 87 | """ |
| 88 | Export the current graph (or a specific repository) to a .cgc bundle. |
| 89 | |
| 90 | Args: |
| 91 | output_path: Path where the .cgc file should be saved |
| 92 | repo_path: Optional specific repository path to export (None = export all) |
| 93 | include_stats: Whether to include detailed statistics |
| 94 | |
| 95 | Returns: |
| 96 | Tuple[bool, str]: (success, message) |
| 97 | """ |
| 98 | try: |
| 99 | info_logger(f"Starting export to {output_path}") |
| 100 | |
| 101 | # Ensure output path has .cgc extension |
| 102 | if not str(output_path).endswith('.cgc'): |
| 103 | output_path = Path(str(output_path) + '.cgc') |
| 104 | |
| 105 | # Create temporary directory for bundle contents |
| 106 | with tempfile.TemporaryDirectory() as temp_dir: |
| 107 | temp_path = Path(temp_dir) |
| 108 | |
| 109 | # Step 1: Extract metadata |
no outgoing calls
no test coverage detected