basic-memory

"""Command module for basic-memory project management.""" import asyncio import os from pathlib import Path import typer from rich.console import Console from rich.table import Table from basic_memory.cli.app import app from basic_memory.cli.commands.cloud import get_authenticated_headers from basic_memory.cli.commands.command_utils import get_project_info from basic_memory.config import ConfigManager import json from datetime import datetime from rich.panel import Panel from basic_memory.mcp.async_client import client from basic_memory.mcp.tools.utils import call_get from basic_memory.schemas.project_info import ProjectList from basic_memory.mcp.tools.utils import call_post from basic_memory.schemas.project_info import ProjectStatusResponse from basic_memory.mcp.tools.utils import call_delete from basic_memory.mcp.tools.utils import call_put from basic_memory.utils import generate_permalink from basic_memory.mcp.tools.utils import call_patch console = Console() # Create a project subcommand project_app = typer.Typer(help="Manage multiple Basic Memory projects") app.add_typer(project_app, name="project") config = ConfigManager().config def format_path(path: str) -> str: """Format a path for display, using ~ for home directory.""" home = str(Path.home()) if path.startswith(home): return path.replace(home, "~", 1) # pragma: no cover return path @project_app.command("list") def list_projects() -> None: """List all Basic Memory projects.""" # Use API to list projects try: auth_headers = {} if config.cloud_mode_enabled: auth_headers = asyncio.run(get_authenticated_headers()) response = asyncio.run(call_get(client, "/projects/projects", headers=auth_headers)) result = ProjectList.model_validate(response.json()) table = Table(title="Basic Memory Projects") table.add_column("Name", style="cyan") table.add_column("Path", style="green") table.add_column("Default", style="magenta") for project in result.projects: is_default = "✓" if project.is_default else "" table.add_row(project.name, format_path(project.path), is_default) console.print(table) except Exception as e: console.print(f"[red]Error listing projects: {str(e)}[/red]") raise typer.Exit(1) if config.cloud_mode_enabled: @project_app.command("add") def add_project_cloud( name: str = typer.Argument(..., help="Name of the project"), set_default: bool = typer.Option(False, "--default", help="Set as default project"), ) -> None: """Add a new project to Basic Memory Cloud""" try: auth_headers = asyncio.run(get_authenticated_headers()) data = {"name": name, "path": generate_permalink(name), "set_default": set_default} response = asyncio.run( call_post(client, "/projects/projects", json=data, headers=auth_headers) ) result = ProjectStatusResponse.model_validate(response.json()) console.print(f"[green]{result.message}[/green]") except Exception as e: console.print(f"[red]Error adding project: {str(e)}[/red]") raise typer.Exit(1) # Display usage hint console.print("\nTo use this project:") console.print(f" basic-memory --project={name} <command>") else: @project_app.command("add") def add_project( name: str = typer.Argument(..., help="Name of the project"), path: str = typer.Argument(..., help="Path to the project directory"), set_default: bool = typer.Option(False, "--default", help="Set as default project"), ) -> None: """Add a new project.""" # Resolve to absolute path resolved_path = Path(os.path.abspath(os.path.expanduser(path))).as_posix() try: data = {"name": name, "path": resolved_path, "set_default": set_default} response = asyncio.run(call_post(client, "/projects/projects", json=data)) result = ProjectStatusResponse.model_validate(response.json()) console.print(f"[green]{result.message}[/green]") except Exception as e: console.print(f"[red]Error adding project: {str(e)}[/red]") raise typer.Exit(1) # Display usage hint console.print("\nTo use this project:") console.print(f" basic-memory --project={name} <command>") @project_app.command("remove") def remove_project( name: str = typer.Argument(..., help="Name of the project to remove"), ) -> None: """Remove a project.""" try: auth_headers = {} if config.cloud_mode_enabled: auth_headers = asyncio.run(get_authenticated_headers()) project_permalink = generate_permalink(name) response = asyncio.run( call_delete(client, f"/projects/{project_permalink}", headers=auth_headers) ) result = ProjectStatusResponse.model_validate(response.json()) console.print(f"[green]{result.message}[/green]") except Exception as e: console.print(f"[red]Error removing project: {str(e)}[/red]") raise typer.Exit(1) # Show this message regardless of method used console.print("[yellow]Note: The project files have not been deleted from disk.[/yellow]") if not config.cloud_mode_enabled: @project_app.command("default") def set_default_project( name: str = typer.Argument(..., help="Name of the project to set as CLI default"), ) -> None: """Set the default project when 'config.default_project_mode' is set.""" try: project_permalink = generate_permalink(name) response = asyncio.run(call_put(client, f"/projects/{project_permalink}/default")) result = ProjectStatusResponse.model_validate(response.json()) console.print(f"[green]{result.message}[/green]") except Exception as e: console.print(f"[red]Error setting default project: {str(e)}[/red]") raise typer.Exit(1) @project_app.command("sync-config") def synchronize_projects() -> None: """Synchronize project config between configuration file and database.""" # Call the API to synchronize projects try: response = asyncio.run(call_post(client, "/projects/config/sync")) result = ProjectStatusResponse.model_validate(response.json()) console.print(f"[green]{result.message}[/green]") except Exception as e: # pragma: no cover console.print(f"[red]Error synchronizing projects: {str(e)}[/red]") raise typer.Exit(1) @project_app.command("move") def move_project( name: str = typer.Argument(..., help="Name of the project to move"), new_path: str = typer.Argument(..., help="New absolute path for the project"), ) -> None: """Move a project to a new location.""" # Resolve to absolute path resolved_path = Path(os.path.abspath(os.path.expanduser(new_path))).as_posix() try: data = {"path": resolved_path} project_permalink = generate_permalink(name) # TODO fix route to use ProjectPathDep response = asyncio.run( call_patch(client, f"/{name}/project/{project_permalink}", json=data) ) result = ProjectStatusResponse.model_validate(response.json()) console.print(f"[green]{result.message}[/green]") # Show important file movement reminder console.print() # Empty line for spacing console.print( Panel( "[bold red]IMPORTANT:[/bold red] Project configuration updated successfully.\n\n" "[yellow]You must manually move your project files from the old location to:[/yellow]\n" f"[cyan]{resolved_path}[/cyan]\n\n" "[dim]Basic Memory has only updated the configuration - your files remain in their original location.[/dim]", title="⚠️ Manual File Movement Required", border_style="yellow", expand=False, ) ) except Exception as e: console.print(f"[red]Error moving project: {str(e)}[/red]") raise typer.Exit(1) @project_app.command("info") def display_project_info( name: str = typer.Argument(..., help="Name of the project"), json_output: bool = typer.Option(False, "--json", help="Output in JSON format"), ): """Display detailed information and statistics about the current project.""" try: # Get project info info = asyncio.run(get_project_info(name)) if json_output: # Convert to JSON and print print(json.dumps(info.model_dump(), indent=2, default=str)) else: # Project configuration section console.print( Panel( f"Basic Memory version: [bold green]{info.system.version}[/bold green]\n" f"[bold]Project:[/bold] {info.project_name}\n" f"[bold]Path:[/bold] {info.project_path}\n" f"[bold]Default Project:[/bold] {info.default_project}\n", title="📊 Basic Memory Project Info", expand=False, ) ) # Statistics section stats_table = Table(title="📈 Statistics") stats_table.add_column("Metric", style="cyan") stats_table.add_column("Count", style="green") stats_table.add_row("Entities", str(info.statistics.total_entities)) stats_table.add_row("Observations", str(info.statistics.total_observations)) stats_table.add_row("Relations", str(info.statistics.total_relations)) stats_table.add_row( "Unresolved Relations", str(info.statistics.total_unresolved_relations) ) stats_table.add_row("Isolated Entities", str(info.statistics.isolated_entities)) console.print(stats_table) # Entity types if info.statistics.entity_types: entity_types_table = Table(title="📑 Entity Types") entity_types_table.add_column("Type", style="blue") entity_types_table.add_column("Count", style="green") for entity_type, count in info.statistics.entity_types.items(): entity_types_table.add_row(entity_type, str(count)) console.print(entity_types_table) # Most connected entities if info.statistics.most_connected_entities: # pragma: no cover connected_table = Table(title="🔗 Most Connected Entities") connected_table.add_column("Title", style="blue") connected_table.add_column("Permalink", style="cyan") connected_table.add_column("Relations", style="green") for entity in info.statistics.most_connected_entities: connected_table.add_row( entity["title"], entity["permalink"], str(entity["relation_count"]) ) console.print(connected_table) # Recent activity if info.activity.recently_updated: # pragma: no cover recent_table = Table(title="🕒 Recent Activity") recent_table.add_column("Title", style="blue") recent_table.add_column("Type", style="cyan") recent_table.add_column("Last Updated", style="green") for entity in info.activity.recently_updated[:5]: # Show top 5 updated_at = ( datetime.fromisoformat(entity["updated_at"]) if isinstance(entity["updated_at"], str) else entity["updated_at"] ) recent_table.add_row( entity["title"], entity["entity_type"], updated_at.strftime("%Y-%m-%d %H:%M"), ) console.print(recent_table) # Available projects projects_table = Table(title="📁 Available Projects") projects_table.add_column("Name", style="blue") projects_table.add_column("Path", style="cyan") projects_table.add_column("Default", style="green") for name, proj_info in info.available_projects.items(): is_default = name == info.default_project project_path = proj_info["path"] projects_table.add_row(name, project_path, "✓" if is_default else "") console.print(projects_table) # Timestamp current_time = ( datetime.fromisoformat(str(info.system.timestamp)) if isinstance(info.system.timestamp, str) else info.system.timestamp ) console.print(f"\nTimestamp: [cyan]{current_time.strftime('%Y-%m-%d %H:%M:%S')}[/cyan]") except Exception as e: # pragma: no cover typer.echo(f"Error getting project info: {e}", err=True) raise typer.Exit(1)