Function clean_helper

src/codegraphcontext/cli/cli_helpers.py:556–601

Remove orphaned nodes and relationships from the database.

(context: Optional[str] = None)

def clean_helper(context: Optional[str] = None):
    """Remove orphaned nodes and relationships from the database."""
    services = _initialize_services(context)
    if not all(services[:3]):
        return

    db_manager, _, _, ctx = services
    
    console.print("[cyan]🧹 Cleaning database (removing orphaned nodes)...[/cyan]")
    
    try:
        total_deleted = 0
        batch_size = 500
        
        with db_manager.get_driver().session() as session:
            # Layer-by-layer deletion: iteratively delete nodes that lost
            # their CONTAINS parent. Each pass peels one layer of the
            # Repository → File → Class/Function → Variable hierarchy.
            while True:
                result = session.run("""
                    MATCH (n)
                    WHERE NOT n:Repository
                      AND NOT ()-[:CONTAINS]->(n)
                    WITH n LIMIT $batch_size
                    DETACH DELETE n
                    RETURN count(n) as deleted
                """, batch_size=batch_size)
                record = result.single()
                deleted_count = record["deleted"] if record else 0
                total_deleted += deleted_count
                
                if deleted_count == 0:
                    break
                    
                console.print(f"[dim]  Deleted {deleted_count} orphaned nodes (batch)...[/dim]")
            
            if total_deleted > 0:
                console.print(f"[green]✓[/green] Deleted {total_deleted} orphaned nodes total")
            else:
                console.print("[green]✓[/green] No orphaned nodes found")
            
        console.print("[green]✅ Database cleanup complete![/green]")
    except Exception as e:
        console.print(f"[bold red]An error occurred during cleanup:[/bold red] {e}")
    finally:
        db_manager.close_driver()


def stats_helper(path: str = None, context: Optional[str] = None):