improve error handling and cli message formatting

Author: AlePouroullis

src/humanloop/cli/__main__.py

@@ -5,6 +5,7 @@
 from functools import wraps
 from dotenv import load_dotenv, find_dotenv
 import os
+import sys
 from humanloop import Humanloop
 from humanloop.sync.sync_client import SyncClient
 from datetime import datetime
@@ -18,6 +19,12 @@
 if not logger.hasHandlers():
     logger.addHandler(console_handler)
 
+# Color constants
+SUCCESS_COLOR = "green"
+ERROR_COLOR = "red"
+INFO_COLOR = "blue"
+WARNING_COLOR = "yellow"
+
 def get_client(api_key: Optional[str] = None, env_file: Optional[str] = None, base_url: Optional[str] = None) -> Humanloop:
     """Get a Humanloop client instance."""
     if not api_key:
@@ -36,7 +43,7 @@ def get_client(api_key: Optional[str] = None, env_file: Optional[str] = None, ba
             api_key = os.getenv("HUMANLOOP_API_KEY")
             if not api_key:
                 raise click.ClickException(
-                    "No API key found. Set HUMANLOOP_API_KEY in .env file or environment, or use --api-key"
+                    click.style("No API key found. Set HUMANLOOP_API_KEY in .env file or environment, or use --api-key", fg=ERROR_COLOR)
                 )
 
     return Humanloop(api_key=api_key, base_url=base_url)
@@ -47,21 +54,21 @@ def common_options(f: Callable) -> Callable:
         "--api-key",
         help="Humanloop API key. If not provided, uses HUMANLOOP_API_KEY from .env or environment.",
         default=None,
+        show_default=False,
     )
     @click.option(
         "--env-file",
         help="Path to .env file. If not provided, looks for .env in current directory.",
         default=None,
         type=click.Path(exists=True),
+        show_default=False,
     )
     @click.option(
         "--base-dir",
         help="Base directory for synced files",
         default="humanloop",
         type=click.Path(),
     )
-    # Hidden option for internal use - allows overriding the Humanloop API base URL
-    # Can be set via --base-url or HUMANLOOP_BASE_URL environment variable
     @click.option(
         "--base-url",
         default=None,
@@ -79,20 +86,29 @@ def wrapper(*args, **kwargs):
         try:
             return f(*args, **kwargs)
         except Exception as e:
-            logger.error(f"Error during sync operation: {str(e)}")
-            raise click.ClickException(str(e))
+            click.echo(click.style(str(f"Error: {e}"), fg=ERROR_COLOR))
+            sys.exit(1)
     return wrapper
 
-@click.group()
-def cli():
+@click.group(
+    help="Humanloop CLI for managing sync operations.",
+    context_settings={
+        "help_option_names": ["-h", "--help"],
+        "max_content_width": 100,
+    }
+)
+@common_options
+def cli(api_key: Optional[str], env_file: Optional[str], base_dir: str, base_url: Optional[str]):
     """Humanloop CLI for managing sync operations."""
     pass
 
 @cli.command()
 @click.option(
     "--path",
     "-p",
-    help="Path to pull (file or directory). If not provided, pulls everything.",
+    help="Path to pull (file or directory). If not provided, pulls everything. "+
+    "To pull a specific file, ensure the extension for the file is included (e.g. .prompt or .agent). "+
+    "To pull a directory, simply specify the path to the directory (e.g. abc/def to pull all files under abc/def and its subdirectories).",
     default=None,
 )
 @click.option(
@@ -101,37 +117,37 @@ def cli():
     help="Environment to pull from (e.g. 'production', 'staging')",
     default=None,
 )
-@common_options
 @handle_sync_errors
 def pull(path: Optional[str], environment: Optional[str], api_key: Optional[str], env_file: Optional[str], base_dir: str, base_url: Optional[str]):
-    """Pull files from Humanloop to local filesystem.
-    
-    If PATH is provided and ends with .prompt or .agent, pulls that specific file.
-    Otherwise, pulls all files under the specified directory path.
-    If no PATH is provided, pulls all files from the root.
-    """
+    """Pull files from Humanloop to your local filesystem."""
     client = get_client(api_key, env_file, base_url)
     sync_client = SyncClient(client, base_dir=base_dir)
 
-    click.echo("Pulling files from Humanloop...")
+    click.echo(click.style("Pulling files from Humanloop...", fg=INFO_COLOR))
 
-    click.echo(f"Path: {path or '(root)'}")
-    click.echo(f"Environment: {environment or '(default)'}")
+    click.echo(click.style(f"Path: {path or '(root)'}", fg=INFO_COLOR))
+    click.echo(click.style(f"Environment: {environment or '(default)'}", fg=INFO_COLOR))
 
     successful_files = sync_client.pull(path, environment)
 
     # Get metadata about the operation
     metadata = sync_client.metadata.get_last_operation()
     if metadata:
-        click.echo(f"\nSync completed in {metadata['duration_ms']}ms")
+        # Determine if the operation was successful based on failed_files
+        is_successful = not metadata.get('failed_files') and not metadata.get('error')
+        duration_color = SUCCESS_COLOR if is_successful else ERROR_COLOR
+        click.echo(click.style(f"\nSync completed in {metadata['duration_ms']}ms", fg=duration_color))
+        
         if metadata['successful_files']:
-            click.echo(f"\nSuccessfully synced {len(metadata['successful_files'])} files:")
+            click.echo(click.style(f"\nSuccessfully synced {len(metadata['successful_files'])} files:", fg=SUCCESS_COLOR))
             for file in metadata['successful_files']:
-                click.echo(f"  ✓ {file}")
+                click.echo(click.style(f"  ✓ {file}", fg=SUCCESS_COLOR))
         if metadata['failed_files']:
-            click.echo(f"\nFailed to sync {len(metadata['failed_files'])} files:")
+            click.echo(click.style(f"\nFailed to sync {len(metadata['failed_files'])} files:", fg=ERROR_COLOR))
             for file in metadata['failed_files']:
-                click.echo(f"  ✗ {file}")
+                click.echo(click.style(f"  ✗ {file}", fg=ERROR_COLOR))
+        if metadata.get('error'):
+            click.echo(click.style(f"\nError: {metadata['error']}", fg=ERROR_COLOR))
 
 def format_timestamp(timestamp: str) -> str:
     """Format timestamp to a more readable format."""
@@ -147,7 +163,6 @@ def format_timestamp(timestamp: str) -> str:
     is_flag=True,
     help="Display history in a single line per operation",
 )
-@common_options
 @handle_sync_errors
 def history(api_key: Optional[str], env_file: Optional[str], base_dir: str, base_url: Optional[str], oneline: bool):
     """Show sync operation history."""
@@ -156,32 +171,32 @@ def history(api_key: Optional[str], env_file: Optional[str], base_dir: str, base
 
     history = sync_client.metadata.get_history()
     if not history:
-        click.echo("No sync operations found in history.")
+        click.echo(click.style("No sync operations found in history.", fg=WARNING_COLOR))
         return
 
     if not oneline:
-        click.echo("Sync Operation History:")
-        click.echo("======================")
+        click.echo(click.style("Sync Operation History:", fg=INFO_COLOR))
+        click.echo(click.style("======================", fg=INFO_COLOR))
 
     for op in history:
         if oneline:
             # Format: timestamp | operation_type | path | environment | duration_ms | status
-            status = "✓" if not op['failed_files'] else "✗"
+            status = click.style("✓", fg=SUCCESS_COLOR) if not op['failed_files'] else click.style("✗", fg=ERROR_COLOR)
             click.echo(f"{format_timestamp(op['timestamp'])} | {op['operation_type']} | {op['path'] or '(root)'} | {op['environment'] or '-'} | {op['duration_ms']}ms | {status}")
         else:
-            click.echo(f"\nOperation: {op['operation_type']}")
+            click.echo(click.style(f"\nOperation: {op['operation_type']}", fg=INFO_COLOR))
             click.echo(f"Timestamp: {format_timestamp(op['timestamp'])}")
             click.echo(f"Path: {op['path'] or '(root)'}")
             if op['environment']:
                 click.echo(f"Environment: {op['environment']}")
             click.echo(f"Duration: {op['duration_ms']}ms")
             if op['successful_files']:
-                click.echo(f"Successfully synced {len(op['successful_files'])} file{'' if len(op['successful_files']) == 1 else 's'}")
+                click.echo(click.style(f"Successfully synced {len(op['successful_files'])} file{'' if len(op['successful_files']) == 1 else 's'}", fg=SUCCESS_COLOR))
             if op['failed_files']:
-                click.echo(f"Failed to sync {len(op['failed_files'])} file{'' if len(op['failed_files']) == 1 else 's'}")
+                click.echo(click.style(f"Failed to sync {len(op['failed_files'])} file{'' if len(op['failed_files']) == 1 else 's'}", fg=ERROR_COLOR))
             if op['error']:
-                click.echo(f"Error: {op['error']}")
-            click.echo("----------------------")
+                click.echo(click.style(f"Error: {op['error']}", fg=ERROR_COLOR))
+            click.echo(click.style("----------------------", fg=INFO_COLOR))
 
 if __name__ == "__main__":
     cli() 

src/humanloop/overload.py

@@ -1,14 +1,15 @@
 import inspect
 import logging
 import types
+import warnings
 from typing import TypeVar, Union, Literal, Optional
 from pathlib import Path
 from humanloop.context import (
     get_decorator_context,
     get_evaluation_context,
     get_trace_id,
 )
-from humanloop.evals.run import HumanloopRuntimeError
+from humanloop.error import HumanloopRuntimeError
 
 from humanloop.evaluators.client import EvaluatorsClient
 from humanloop.flows.client import FlowsClient
@@ -154,34 +155,35 @@ def overload_with_local_files(
         use_local_files: Whether to use local files
         
     Raises:
-        FileNotFoundError: If use_local_files is True and local file is not found
-        IOError: If use_local_files is True and local file cannot be read
+        HumanloopRuntimeError: If use_local_files is True and local file cannot be accessed
     """
     original_call = client._call if hasattr(client, '_call') else client.call
     original_log = client._log if hasattr(client, '_log') else client.log
     file_type = _get_file_type_from_client(client)
 
     def _overload(self, function_name: str, **kwargs) -> PromptCallResponse:
+        if "id" and "path" in kwargs: 
+            raise HumanloopRuntimeError(f"Can only specify one of `id` or `path` when {function_name}ing a {file_type}")
         # Handle local files if enabled
         if use_local_files and "path" in kwargs:
             # Check if version_id or environment is specified
             has_version_info = "version_id" in kwargs or "environment" in kwargs
 
             if has_version_info:
-                logger.warning(
-                    "Ignoring local file for %s as version_id or environment was specified. "
+                warnings.warn(
+                    f"Ignoring local file for {kwargs['path']} as version_id or environment was specified. "
                     "Using remote version instead.",
-                    kwargs["path"]
+                    UserWarning
                 )
             else:
                 # Only use local file if no version info is specified
                 normalized_path = sync_client._normalize_path(kwargs["path"])
                 try:
                     file_content = sync_client.get_file_content(normalized_path, file_type)
                     kwargs[file_type] = file_content
-                except (FileNotFoundError, IOError) as e:
+                except (HumanloopRuntimeError) as e:
                     # Re-raise with more context
-                    raise type(e)(f"Failed to use local file for {kwargs['path']}: {str(e)}")
+                    raise HumanloopRuntimeError(f"Failed to use local file for {kwargs['path']}: {str(e)}")
 
         try:
             if function_name == "call":

src/humanloop/sync/sync_client.py

@@ -5,6 +5,7 @@
 from humanloop.types import FileType
 from .metadata_handler import MetadataHandler
 import time
+from humanloop.error import HumanloopRuntimeError
 
 if TYPE_CHECKING:
     from humanloop.base_client import BaseHumanloop
@@ -67,16 +68,15 @@ def _get_file_content_impl(self, path: str, file_type: FileType) -> str:
             The file content
             
         Raises:
-            FileNotFoundError: If the file doesn't exist
-            IOError: If there's an error reading the file
+            HumanloopRuntimeError: If the file doesn't exist or can't be read
         """
         # Construct path to local file
         local_path = self.base_dir / path
         # Add appropriate extension
         local_path = local_path.parent / f"{local_path.stem}.{file_type}"
 
         if not local_path.exists():
-            raise FileNotFoundError(f"Local file not found: {local_path}")
+            raise HumanloopRuntimeError(f"Local file not found: {local_path}")
 
         try:
             # Read the file content
@@ -85,7 +85,7 @@ def _get_file_content_impl(self, path: str, file_type: FileType) -> str:
             logger.debug(f"Using local file content from {local_path}")
             return file_content
         except Exception as e:
-            raise IOError(f"Error reading local file {local_path}: {str(e)}")
+            raise HumanloopRuntimeError(f"Error reading local file {local_path}: {str(e)}")
 
     def get_file_content(self, path: str, file_type: FileType) -> str:
         """Get the content of a file from cache or filesystem.
@@ -101,8 +101,7 @@ def get_file_content(self, path: str, file_type: FileType) -> str:
             The file content
             
         Raises:
-            FileNotFoundError: If the file doesn't exist
-            IOError: If there's an error reading the file
+            HumanloopRuntimeError: If the file doesn't exist or can't be read
         """
         return self._get_file_content_impl(path, file_type)
 
@@ -205,7 +204,7 @@ def _pull_file(self, path: str, environment: str | None = None) -> None:
         self._save_serialized_file(file.content, file.path, file.type)
 
     def _pull_directory(self, 
-            path: str | None = None,    
+            directory: str | None = None,    
             environment: str | None = None, 
         ) -> List[str]:
         """Sync prompt and agent files from Humanloop to local filesystem.
@@ -219,6 +218,9 @@ def _pull_directory(self,
 
         Returns:
             List of successfully processed file paths
+
+        Raises:
+            Exception: If there is an error fetching files from Humanloop
         """
         successful_files = []
         failed_files = []
@@ -231,7 +233,7 @@ def _pull_directory(self,
                     page=page,
                     include_content=True,
                     environment=environment,
-                    directory=path
+                    directory=directory
                 )
 
                 if len(response.records) == 0:
@@ -258,14 +260,14 @@ def _pull_directory(self,
 
                 page += 1
             except Exception as e:
-                logger.error(f"Failed to fetch page {page}: {str(e)}")
-                break
+                raise HumanloopRuntimeError(f"Failed to fetch page {page}: {str(e)}")
 
-        # Log summary
-        if successful_files:
-            logger.info(f"\nSynced {len(successful_files)} files")
-        if failed_files:
-            logger.error(f"Failed to sync {len(failed_files)} files")
+        # Log summary only if we have results
+        if successful_files or failed_files:
+            if successful_files:
+                logger.info(f"\nSynced {len(successful_files)} files")
+            if failed_files:
+                logger.error(f"Failed to sync {len(failed_files)} files")
 
         return successful_files
 
@@ -286,6 +288,7 @@ def pull(self, path: str | None = None, environment: str | None = None) -> List[
         start_time = time.time()
         try:
             if path is None:
+                # Pull all files from the root
                 successful_files = self._pull_directory(None, environment)
                 failed_files = []  # Failed files are already logged in _pull_directory
             else:

tests/conftest.py

@@ -196,7 +196,7 @@ def humanloop_client(request, api_keys: APIKeys) -> Humanloop:
     use_local_files = getattr(request, "param", False)
     return Humanloop(
         api_key=api_keys.humanloop,
-        base_url="http://localhost:80/v5/",
+        base_url="http://localhost:80/v5",
         use_local_files=use_local_files
     )