Helpers

provide.foundation.cli.helpers

TODO: Add module docstring.

Classes

Functions

build_attributes_from_args

build_attributes_from_args(
    json_attrs: str | None, attr: tuple[str, ...]
) -> tuple[dict[str, Any], int]

Build attributes dictionary from JSON and key=value arguments.

Uses the parsers module for consistent type inference and parsing.

Parameters:

Name	Type	Description	Default
json_attrs	str | None	JSON string of attributes	required
attr	tuple[str, ...]	Tuple of key=value attribute strings	required

Returns:

Type	Description
tuple[dict[str, Any], int]	Tuple of (attributes dict, error_code). Error code is 0 on success.

Source code in provide/foundation/cli/helpers.py

def build_attributes_from_args(
    json_attrs: str | None,
    attr: tuple[str, ...],
) -> tuple[dict[str, Any], int]:
    """Build attributes dictionary from JSON and key=value arguments.

    Uses the parsers module for consistent type inference and parsing.

    Args:
        json_attrs: JSON string of attributes
        attr: Tuple of key=value attribute strings

    Returns:
        Tuple of (attributes dict, error_code). Error code is 0 on success.

    """
    attributes: dict[str, Any] = {}

    # Parse JSON attributes first
    if json_attrs:
        try:
            json_dict = json_loads(json_attrs)
            if not isinstance(json_dict, dict):
                click.echo("Error: JSON attributes must be an object.", err=True)
                return {}, 1
            attributes.update(json_dict)
        except (ValueError, TypeError, ValidationError) as e:
            click.echo(f"Error: Invalid JSON attributes: {e}", err=True)
            return {}, 1

    # Add key=value attributes with automatic type inference
    for kv_pair in attr:
        try:
            key, value = kv_pair.split("=", 1)
            attributes[key] = _infer_and_parse_value(value)
        except ValueError:
            click.echo(
                f"Error: Invalid attribute format '{kv_pair}'. Use key=value.",
                err=True,
            )
            return {}, 1

    return attributes, 0

format_duration

format_duration(seconds: float) -> str

Format duration in seconds to human-readable string.

Delegates to formatting.numbers.format_duration() with short format and adds spaces between components for CLI readability.

Parameters:

Name	Type	Description	Default
seconds	float	Duration in seconds	required

Returns:

Type	Description
str	Formatted duration string (e.g., "1h 23m 45s", "45s", "1.5s")

Source code in provide/foundation/cli/helpers.py

def format_duration(seconds: float) -> str:
    """Format duration in seconds to human-readable string.

    Delegates to formatting.numbers.format_duration() with short format
    and adds spaces between components for CLI readability.

    Args:
        seconds: Duration in seconds

    Returns:
        Formatted duration string (e.g., "1h 23m 45s", "45s", "1.5s")

    """
    # Handle sub-minute durations with decimal precision
    if seconds < 60:
        return f"{seconds:.1f}s"

    # Use formatting module for consistency, with short format
    # The formatting module produces "1h23m45s", we want "1h 23m 45s"
    formatted = _format_duration(seconds, short=True)

    # Add spaces between components for better CLI readability
    # Transform "1h23m45s" → "1h 23m 45s"
    result = formatted.replace("h", "h ").replace("m", "m ").replace("d", "d ")
    return result.rstrip()  # Remove trailing space

get_client_from_context

get_client_from_context(ctx: Any) -> tuple[Any | None, int]

Get OpenObserve client from Click context.

Parameters:

Name	Type	Description	Default
ctx	Any	Click context object	required

Returns:

Type	Description
tuple[Any | None, int]	Tuple of (client, error_code). Error code is 0 on success.

Source code in provide/foundation/cli/helpers.py

def get_client_from_context(ctx: Any) -> tuple[Any | None, int]:
    """Get OpenObserve client from Click context.

    Args:
        ctx: Click context object

    Returns:
        Tuple of (client, error_code). Error code is 0 on success.

    """
    client = ctx.obj.get("client") if ctx.obj else None
    if not client:
        click.echo("Error: OpenObserve not configured.", err=True)
        return None, 1
    return client, 0

get_message_from_stdin

get_message_from_stdin() -> tuple[str | None, int]

Get message from stdin if available.

Returns:

Type	Description
str | None	Tuple of (message, error_code). If successful, error_code is 0.
int	If stdin is a TTY (no piped input), returns (None, 1).

Source code in provide/foundation/cli/helpers.py

def get_message_from_stdin() -> tuple[str | None, int]:
    """Get message from stdin if available.

    Returns:
        Tuple of (message, error_code). If successful, error_code is 0.
        If stdin is a TTY (no piped input), returns (None, 1).

    """
    if sys.stdin.isatty():
        return None, 1

    try:
        message = sys.stdin.read().strip()
        if not message:
            click.echo("Error: Empty input from stdin.", err=True)
            return None, 1
        return message, 0
    except Exception as e:
        click.echo(f"Error reading from stdin: {e}", err=True)
        return None, 1

parse_filter_string

parse_filter_string(filter_str: str) -> dict[str, str]

Parse filter string into key-value dictionary.

Uses the parsers module for consistent parsing behavior.

Parameters:

Name	Type	Description	Default
filter_str	str	Filter string in format 'key1=value1,key2=value2'	required

Returns:

Type	Description
dict[str, str]	Dictionary of filter key-value pairs

Source code in provide/foundation/cli/helpers.py

def parse_filter_string(filter_str: str) -> dict[str, str]:
    """Parse filter string into key-value dictionary.

    Uses the parsers module for consistent parsing behavior.

    Args:
        filter_str: Filter string in format 'key1=value1,key2=value2'

    Returns:
        Dictionary of filter key-value pairs

    """
    if not filter_str:
        return {}

    try:
        return parse_dict(filter_str, item_separator=",", key_separator="=", strip=True)
    except ValueError as e:
        click.echo(f"Warning: Invalid filter format: {e}", err=True)
        return {}

requires_click

requires_click(func: Callable[P, R]) -> Callable[P, R]

Decorator to ensure Click is available for CLI commands.

Replaces the boilerplate if _HAS_CLICK / else ImportError stub pattern.

Example

@requires_click def my_command(*args, **kwargs): # Command implementation pass

Parameters:

Name	Type	Description	Default
func	Callable[P, R]	CLI command function to wrap	required

Returns:

Type	Description
Callable[P, R]	Wrapped function that raises ImportError if Click is not available

Source code in provide/foundation/cli/helpers.py

def requires_click(func: Callable[P, R]) -> Callable[P, R]:
    """Decorator to ensure Click is available for CLI commands.

    Replaces the boilerplate if _HAS_CLICK / else ImportError stub pattern.

    Example:
        @requires_click
        def my_command(*args, **kwargs):
            # Command implementation
            pass

    Args:
        func: CLI command function to wrap

    Returns:
        Wrapped function that raises ImportError if Click is not available

    """

    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        if not _HAS_CLICK:
            raise ImportError(
                "CLI commands require optional dependencies. "
                "Install with: pip install 'provide-foundation[cli]'"
            )
        return func(*args, **kwargs)

    return wrapper