Something went wrong. Try again.
Command line interface for accessing the Open Web Index shards and slices.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160"""Command Executor - Wraps command execution with logging, timing, and error handling.
This module provides a standardized way to execute CLI commands with:- Timing measurement- Error categorization- Logging to both console and file- Structured result output for JSON mode"""import timeimport traceback as tbfrom typing import Any, Callable, Optional
from owilix.core.types import CommandResult, ErrorType, ExitCode
class CommandExecutor: """ Executor that wraps command functions with standard handling. Features: - Automatic timing - Exception catching and categorization - Logging to OWI logfile - Structured output for piping """ def __init__(self, ctx): """ Initialize executor with CLI context. Args: ctx: CLIContext with owi, console, etc. """ self.ctx = ctx def execute( self, func: Callable, *args, command_name: str = "", **kwargs ) -> CommandResult: """ Execute a command function with standard wrapping. Args: func: The command function to execute *args: Positional arguments for the function command_name: Name of the command (for logging) **kwargs: Keyword arguments for the function Returns: CommandResult with timing, success status, and any errors """ command_name = command_name or func.__name__ start_time = time.time() result = CommandResult(command=command_name) try: # Execute the actual command return_value = func(*args, **kwargs) # Handle different return types if isinstance(return_value, CommandResult): result = return_value result.command = command_name else: result.success = True result.object = return_value result.exit_code = ExitCode.SUCCESS except ConnectionError as e: result = self._handle_error(e, ErrorType.NETWORK, ExitCode.NETWORK_ERROR, command_name) except TimeoutError as e: result = self._handle_error(e, ErrorType.NETWORK, ExitCode.NETWORK_ERROR, command_name) except PermissionError as e: result = self._handle_error(e, ErrorType.AUTH, ExitCode.AUTH_ERROR, command_name) except ValueError as e: result = self._handle_error(e, ErrorType.VALIDATION, ExitCode.VALIDATION_ERROR, command_name) except FileNotFoundError as e: result = self._handle_error(e, ErrorType.DATA, ExitCode.DATA_ERROR, command_name) except OSError as e: result = self._handle_error(e, ErrorType.DATA, ExitCode.DATA_ERROR, command_name) except Exception as e: # Print to console so errors aren't silent if hasattr(self.ctx, 'console'): self.ctx.console.print(f"[red]Error executing {command_name}: {e}[/red]") self.ctx.console.print(f"[dim]{tb.format_exc()}[/dim]") result = self._handle_error(e, ErrorType.INTERNAL, ExitCode.ERROR, command_name) finally: result.duration_ms = (time.time() - start_time) * 1000 self._log_execution(result) return result def _handle_error( self, error: Exception, error_type: str, exit_code: int, command_name: str ) -> CommandResult: """Create error result from exception.""" return CommandResult.error_result( msg=str(error), error_type=error_type, exit_code=exit_code, traceback=tb.format_exc(), command=command_name, ) def _log_execution(self, result: CommandResult) -> None: """Log command execution to OWI log file.""" if hasattr(self.ctx, 'owi') and hasattr(self.ctx.owi, 'update_log'): log_entry = { "command": result.command, "success": result.success, "duration_ms": round(result.duration_ms, 2), "exit_code": result.exit_code, } if not result.success: log_entry["error_type"] = result.error_type log_entry["error_msg"] = result.msg[:200] # Truncate long errors try: self.ctx.owi.update_log(log_entry) except Exception: pass # Don't fail on logging errors
def execute_command( ctx, func: Callable, *args, command_name: str = "", **kwargs) -> CommandResult: """ Convenience function to execute a command with standard handling. Args: ctx: CLIContext func: Command function to execute *args: Positional arguments command_name: Name for logging **kwargs: Keyword arguments Returns: CommandResult Example: result = execute_command(ctx, do_remote_ls, specifier, command_name="remote ls") if not result.success: raise typer.Exit(code=result.exit_code) """ executor = CommandExecutor(ctx) return executor.execute(func, *args, command_name=command_name, **kwargs)