Function reindex_helper

src/codegraphcontext/cli/cli_helpers.py:672–715

Force re-index by deleting and rebuilding the repository.

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

def reindex_helper(path: str, context: Optional[str] = None):
    """Force re-index by deleting and rebuilding the repository."""
    time_start = time.time()
    path_obj = Path(path).resolve()
    index_cwd = path_obj if path_obj.is_dir() else path_obj.parent
    services = _initialize_services(context, cwd=index_cwd)
    if not all(services[:3]):
        _fail_services_init()

    db_manager, graph_builder, code_finder, ctx = services

    if not path_obj.exists():
        console.print(f"[red]Error: Path does not exist: {path_obj}[/red]")
        db_manager.close_driver()
        raise typer.Exit(code=1)

    # Check if already indexed
    indexed_repos = code_finder.list_indexed_repositories()
    repo_exists = any_repo_matches_path(indexed_repos, path_obj)

    if repo_exists:
        console.print(f"[yellow]Deleting existing index for: {path_obj}[/yellow]")
        try:
            graph_builder.delete_repository_from_graph(str(path_obj))
            console.print("[green]✓[/green] Deleted old index")
        except Exception as e:
            console.print(f"[red]Error deleting old index: {e}[/red]")
            db_manager.close_driver()
            raise typer.Exit(code=1)
    
    console.print(f"[cyan]Re-indexing: {path_obj}[/cyan]")
    
    try:
        asyncio.run(_run_index_with_progress(graph_builder, path_obj, is_dependency=False, cgcignore_path=ctx.cgcignore_path))
        time_end = time.time()
        elapsed = time_end - time_start
        _print_call_resolution_diagnostics(graph_builder)
        _print_index_execution_summary(graph_builder)
        console.print(f"[green]Successfully re-indexed: {path} in {elapsed:.2f} seconds[/green]")
    except Exception as e:
        console.print(f"[bold red]An error occurred during re-indexing:[/bold red] {e}")
        raise typer.Exit(code=1)
    finally:
        db_manager.close_driver()


def update_helper(path: str, context: Optional[str] = None, quiet: bool = False):