cloud_utils.py

"""Shared utilities for cloud operations."""

from basic_memory.cli.commands.cloud.api_client import make_api_request
from basic_memory.config import ConfigManager
from basic_memory.schemas.cloud import (
    CloudProjectList,
    CloudProjectCreateRequest,
    CloudProjectCreateResponse,
)
from basic_memory.utils import generate_permalink


class CloudUtilsError(Exception):
    """Exception raised for cloud utility errors."""

    pass


def _workspace_headers(workspace: str | None = None) -> dict[str, str]:
    """Build workspace header if workspace is specified."""
    if workspace:
        return {"X-Workspace-ID": workspace}
    return {}


async def fetch_cloud_projects(
    *,
    workspace: str | None = None,
    api_request=make_api_request,
) -> CloudProjectList:
    """Fetch list of projects from cloud API.

    Args:
        workspace: Cloud workspace tenant_id to list projects from

    Returns:
        CloudProjectList with projects from cloud
    """
    try:
        config_manager = ConfigManager()
        config = config_manager.config
        host_url = config.cloud_host.rstrip("/")

        response = await api_request(
            method="GET",
            url=f"{host_url}/proxy/v2/projects/",
            headers=_workspace_headers(workspace),
        )

        return CloudProjectList.model_validate(response.json())
    except Exception as e:
        raise CloudUtilsError(f"Failed to fetch cloud projects: {e}") from e


async def create_cloud_project(
    project_name: str,
    *,
    workspace: str | None = None,
    api_request=make_api_request,
) -> CloudProjectCreateResponse:
    """Create a new project on cloud.

    Args:
        project_name: Name of project to create
        workspace: Cloud workspace tenant_id to create project in

    Returns:
        CloudProjectCreateResponse with project details from API
    """
    try:
        config_manager = ConfigManager()
        config = config_manager.config
        host_url = config.cloud_host.rstrip("/")

        # Use generate_permalink to ensure consistent naming
        project_path = generate_permalink(project_name)

        project_data = CloudProjectCreateRequest(
            name=project_name,
            path=project_path,
            set_default=False,
        )

        headers = {"Content-Type": "application/json"}
        headers.update(_workspace_headers(workspace))

        response = await api_request(
            method="POST",
            url=f"{host_url}/proxy/v2/projects/",
            headers=headers,
            json_data=project_data.model_dump(),
        )

        return CloudProjectCreateResponse.model_validate(response.json())
    except Exception as e:
        raise CloudUtilsError(f"Failed to create cloud project '{project_name}': {e}") from e


async def sync_project(project_name: str, force_full: bool = False) -> None:
    """Trigger sync for a specific project on cloud.

    Args:
        project_name: Name of project to sync
        force_full: If True, force a full scan bypassing watermark optimization
    """
    try:
        from basic_memory.cli.commands.command_utils import run_sync

        await run_sync(project=project_name, force_full=force_full)
    except Exception as e:
        raise CloudUtilsError(f"Failed to sync project '{project_name}': {e}") from e


async def project_exists(
    project_name: str,
    *,
    workspace: str | None = None,
    api_request=make_api_request,
) -> bool:
    """Check if a project exists on cloud.

    Args:
        project_name: Name of project to check
        workspace: Cloud workspace tenant_id to check in

    Returns:
        True if project exists, False otherwise
    """
    try:
        projects = await fetch_cloud_projects(workspace=workspace, api_request=api_request)
        project_names = {p.name for p in projects.projects}
        return project_name in project_names
    except Exception:
        return False