Something went wrong. Try again.
Command line interface for accessing the Open Web Index shards and slices.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118"""Plugin Commands - Typer CLI for loading and executing external plugins.
This enables dynamic loading of plugin classes that extend BaseCommand."""import importlibimport astfrom typing import List, Optionalimport typer
from ._common.context import CLIContext
# Create sub-app for plugin commandsapp = typer.Typer( name="plugin", help="Load and execute external plugin commands", no_args_is_help=True,)
@app.command("run")def run_plugin( ctx: typer.Context, cmd_class: str = typer.Argument(..., help="Full class path (e.g., owilix.plugins.push.opensearch.OpenSearchIndexer)"), command: str = typer.Argument(..., help="Subcommand to execute (e.g., 'index')"), args: Optional[List[str]] = typer.Argument(None, help="Additional arguments for the plugin command"),): """ Execute a plugin command dynamically. Loads a plugin class and executes its subcommand with provided arguments. Examples: owi plugin run owilix.plugins.push.opensearch.OpenSearchIndexer index my_index owi plugin run owilix.plugins.ngram.BloomNGrams create "all:latest/collectionName=main" """ cli_ctx: CLIContext = ctx.obj try: # Split the class path into module and class name module_path, class_name = cmd_class.rsplit('.', 1) # Import the module module = importlib.import_module(module_path) plugin_class = getattr(module, class_name) # Build kwargs from args (key=value pattern) cmd_args = [] cmd_kwargs = {} if args: for arg in args: if '=' in arg and not arg.startswith('('): key, value = arg.split('=', 1) # Try to parse as Python literals try: cmd_kwargs[key] = ast.literal_eval(value) except: cmd_kwargs[key] = value else: # Strip parentheses if present if arg.startswith('(') and arg.endswith(')'): arg = arg[1:-1] cmd_args.append(arg) # Instantiate the plugin with proper kwargs plugin_instance = plugin_class( cli_ctx.owi, CONSOLE=cli_ctx.console, TARGET=cli_ctx.target, AUTOYES=cli_ctx.auto_yes, owilix_config=cli_ctx.owi.config, ) # Execute the command plugin_instance(command, *cmd_args, **cmd_kwargs) except ValueError as e: cli_ctx.console.print(f"[red]Invalid class path:[/red] {cmd_class}") cli_ctx.console.print(f"[dim]Expected format: module.submodule.ClassName[/dim]") raise typer.Exit(code=1) except ModuleNotFoundError as e: cli_ctx.console.print(f"[red]Module not found:[/red] {e}") raise typer.Exit(code=1) except AttributeError as e: cli_ctx.console.print(f"[red]Class not found:[/red] {e}") raise typer.Exit(code=1) except Exception as e: cli_ctx.console.print(f"[red]Plugin error:[/red] {e}") raise typer.Exit(code=1)
@app.command("list")def list_plugins(ctx: typer.Context): """ List available plugins. Shows discovered plugin classes from the owilix.plugins package. """ cli_ctx: CLIContext = ctx.obj cli_ctx.console.print("[bold]Available Plugins:[/bold]\n") plugins = [ ("owilix.plugins.push.opensearch.OpenSearchIndexer", "Push data to OpenSearch", ["index"]), ("owilix.plugins.ngram.BloomNGrams", "Bloom filter n-gram operations", ["create", "query"]), ("owilix.plugins.search_consumers.OpenSearchConsumer", "Stream consumer for OpenSearch", []), ] for class_path, description, commands in plugins: cli_ctx.console.print(f" [cyan]{class_path}[/cyan]") cli_ctx.console.print(f" {description}") if commands: cli_ctx.console.print(f" Commands: {', '.join(commands)}") cli_ctx.console.print() cli_ctx.console.print("[dim]Use 'owi plugin run <class_path> <command> [args...]' to execute[/dim]")