Something went wrong. Try again.
Command line interface for accessing the Open Web Index shards and slices.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127"""CLI Context - Pydantic-based context for all CLI commands.
Provides strongly-typed configuration for the CLI, including:- OWIlixManager instance- Console for output- Output format preferences- Logging settings"""from typing import Optional, Anyfrom pydantic import BaseModel, Field, ConfigDictfrom rich.console import Consoleimport loggingimport os
class CLIContext(BaseModel): """ Strongly-typed CLI context passed to all commands. Attributes: owi: OWIlixManager instance for dataset operations console: Rich console for formatted output verbose: Enable verbose output logging output_format: Output format (table, json, jsonl) output_file: Optional file path for output (None = stdout) no_display: Suppress console output (for piping) target: Target directory for local data storage """ model_config = ConfigDict(arbitrary_types_allowed=True) owi: Any = Field(default=None, description="OWIlixManager instance") console: Console = Field(default_factory=Console, description="Rich console for output") verbose: bool = Field(default=False, description="Enable verbose output") output_format: str = Field(default="table", description="Output format: table, json, jsonl") output_file: Optional[str] = Field(default=None, description="Write output to file") no_display: bool = Field(default=False, description="Suppress console output") target: Optional[str] = Field(default=None, description="Target directory") no_progress: bool = Field(default=False, description="Suppress progress bars") fields: Optional[str] = Field(default=None, description="Custom fields (+field,-field)") sql_log: bool = Field(default=False, description="Write executed SQL to sql.log") # Additional context from legacy CLI showfields: Optional[str] = Field(default=None, description="Fields to show in display") nodisplay: Optional[str] = Field(default=None, description="Fields to exclude") auto_yes: bool = Field(default=False, description="Auto-confirm prompts")
def create_context( verbose: bool = False, config_path: Optional[str] = None, output_format: str = "table", output_file: Optional[str] = None, no_display: bool = False, target: Optional[str] = None, no_progress: bool = False, fields: Optional[str] = None, autoyes: bool = False, loglevel: Optional[str] = None, no_remote_config: bool = False, sql_log: bool = False,) -> CLIContext: """ Create a CLI context with initialized OWIlixManager. Args: verbose: Enable verbose logging config_path: Path to configuration file output_format: Output format (table, json, jsonl) output_file: Optional output file path no_display: Suppress console output target: Target directory for data no_progress: Suppress progress bars fields: Custom field selection autoyes: Auto-confirm prompts loglevel: Log level (DEBUG, INFO, WARNING, ERROR) Returns: Initialized CLIContext """ from owilix.core import OWIlixManager from owilix.core.manager import OWIlixConfig, OWILIXEnv from owilix.core.manager.ui import OWILIXConsole from owilix.core.sql_logging import configure_sql_logging # Determine target path owi_path = target or OWILIXEnv.values.owi_path # Load configuration config_file = config_path or os.path.join(owi_path, "owilix.cfg") config = OWIlixConfig(config_file, no_remote=no_remote_config) # Configure logging based on loglevel or verbose if loglevel and loglevel.upper() in ("DEBUG", "INFO", "WARNING", "ERROR"): log_level = getattr(logging, loglevel.upper()) else: log_level = logging.DEBUG if verbose else logging.WARNING logger = logging.getLogger("owilix") logger.setLevel(log_level) # Create OWILIXConsole for unified console+logging console = OWILIXConsole(logger=logger, default_level=log_level) # Create manager with owi_path and config owi = OWIlixManager(owi_path=owi_path, config=config) configure_sql_logging(sql_log, owi.sql_logfile) return CLIContext( owi=owi, console=console, verbose=verbose, output_format=output_format, output_file=output_file, no_display=no_display, target=owi_path, no_progress=no_progress, fields=fields, auto_yes=autoyes, sql_log=sql_log, )
def get_context(ctx) -> CLIContext: """Get CLIContext from Typer context.""" return ctx.obj