Something went wrong. Try again.
Command line interface for accessing the Open Web Index shards and slices.
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677"""Progress utilities for CLI commands.
Re-exports ui.py progress bar helpers for consistent usage across CLI."""from typing import Optional, Anyfrom contextlib import contextmanager
from owilix.core.manager.ui import EnhancedProgressDisplay, ErrorCollector
# Import simple progress from base.py for simpler operationsfrom rich.progress import Progress, BarColumn, TextColumn, TimeRemainingColumn
class DummyProgress: """A no-op progress bar compatible with Rich Progress interface.""" def __init__(self, *args, **kwargs): pass def __enter__(self): return self def __exit__(self, *args, **kwargs): pass def add_task(self, *args, **kwargs): return 0 def update(self, *args, **kwargs): pass def start(self, *args, **kwargs): pass def stop(self, *args, **kwargs): pass
def simple_progress(ctx: Optional[Any] = None) -> Progress: """ Create a simple progress bar for operations like ls, pull, push. Args: ctx: Optional CLIContext. If provided and ctx.no_progress is True, returns DummyProgress. Returns: Rich Progress context manager or DummyProgress """ if ctx and getattr(ctx, 'no_progress', False): return DummyProgress() return Progress( TextColumn("[progress.description]{task.description}"), BarColumn(), "[progress.percentage]{task.percentage:>3.0f}%", TimeRemainingColumn(), TextColumn("{task.fields[current_item]}"), )
def create_enhanced_progress(console, task_name: str = "records processed", ctx: Optional[Any] = None): """ Create an enhanced progress display for complex operations with error tracking. Args: console: OWILIXConsole or Rich Console task_name: Description of what's being counted ctx: Optional CLIContext. If provided and ctx.no_progress is True, returns dummies. Returns: tuple: (EnhancedProgressDisplay/Dummy, ErrorCollector) """ error_collector = ErrorCollector(max_recent_errors=10) if ctx and getattr(ctx, 'no_progress', False): return DummyProgress(), error_collector progress = EnhancedProgressDisplay( console, error_collector, task_name=task_name ) return progress, error_collector
__all__ = [ "simple_progress", "create_enhanced_progress", "EnhancedProgressDisplay", "ErrorCollector",]