Formatters

provide.foundation.integrations.openobserve.formatters

TODO: Add module docstring.

Classes

Functions

format_csv

format_csv(
    response: SearchResponse,
    columns: list[str] | None = None,
) -> str

Format response as CSV.

Parameters:

Name	Type	Description	Default
response	SearchResponse	Search response	required
columns	list[str] | None	Specific columns to include (None for all)	None

Returns:

Type	Description
str	CSV string

Source code in provide/foundation/integrations/openobserve/formatters.py

def format_csv(response: SearchResponse, columns: list[str] | None = None) -> str:
    """Format response as CSV.

    Args:
        response: Search response
        columns: Specific columns to include (None for all)

    Returns:
        CSV string

    """
    if not response.hits:
        return ""

    # Determine columns
    if columns is None:
        all_keys: set[str] = set()
        for hit in response.hits:
            all_keys.update(hit.keys())
        columns = sorted(all_keys)

    # Create CSV
    output = io.StringIO()
    writer = csv.DictWriter(output, fieldnames=columns, extrasaction="ignore")

    writer.writeheader()
    for hit in response.hits:
        # Format timestamp for readability
        if "_timestamp" in hit:
            hit = hit.copy()
            timestamp = hit["_timestamp"]
            if timestamp:
                dt = datetime.fromtimestamp(timestamp / 1_000_000)
                hit["_timestamp"] = dt.isoformat()
        writer.writerow(hit)

    return output.getvalue()

format_json

format_json(
    response: SearchResponse | dict[str, Any],
    pretty: bool = True,
) -> str

Format response as JSON.

Parameters:

Name	Type	Description	Default
response	SearchResponse | dict[str, Any]	Search response or log entry	required
pretty	bool	If True, use pretty printing	True

Returns:

Type	Description
str	JSON string

Source code in provide/foundation/integrations/openobserve/formatters.py

def format_json(response: SearchResponse | dict[str, Any], pretty: bool = True) -> str:
    """Format response as JSON.

    Args:
        response: Search response or log entry
        pretty: If True, use pretty printing

    Returns:
        JSON string

    """
    if isinstance(response, SearchResponse):
        data = {
            "hits": response.hits,
            "total": response.total,
            "took": response.took,
            "scan_size": response.scan_size,
        }
    else:
        data = response

    if pretty:
        return json_dumps(data, indent=2, sort_keys=False)
    return json_dumps(data)

format_log_line

format_log_line(entry: dict[str, Any]) -> str

Format a log entry as a traditional log line.

Parameters:

Name	Type	Description	Default
entry	dict[str, Any]	Log entry dictionary	required

Returns:

Type	Description
str	Formatted log line

Source code in provide/foundation/integrations/openobserve/formatters.py

def format_log_line(entry: dict[str, Any]) -> str:
    """Format a log entry as a traditional log line.

    Args:
        entry: Log entry dictionary

    Returns:
        Formatted log line

    """
    # Extract common fields
    timestamp = entry.get("_timestamp", 0)
    level = entry.get("level", "INFO")
    message = entry.get("message", "")
    service = entry.get("service", "")

    # Convert timestamp to readable format
    if timestamp:
        # Assuming microseconds
        dt = datetime.fromtimestamp(timestamp / 1_000_000)
        time_str = dt.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
    else:
        time_str = "unknown"

    # Build log line
    parts = [time_str, f"[{level:5s}]"]

    if service:
        parts.append(f"[{service}]")

    parts.append(message)

    # Add additional fields as key=value
    exclude_fields = {"_timestamp", "level", "message", "service", "_p"}
    extra_fields = []
    for key, value in entry.items():
        if key not in exclude_fields:
            extra_fields.append(f"{key}={value}")

    if extra_fields:
        parts.append(f"({', '.join(extra_fields)})")

    return " ".join(parts)

format_output

format_output(
    response: SearchResponse | dict[str, Any],
    format_type: str = "log",
    **kwargs: Any
) -> str

Format output based on specified type.

Parameters:

Name	Type	Description	Default
response	SearchResponse | dict[str, Any]	Search response or log entry	required
format_type	str	Output format (json, log, table, csv, summary)	'log'
**kwargs	Any	Additional format-specific options	{}

Returns:

Type	Description
str	Formatted string

Source code in provide/foundation/integrations/openobserve/formatters.py

def format_output(
    response: SearchResponse | dict[str, Any],
    format_type: str = "log",
    **kwargs: Any,
) -> str:
    """Format output based on specified type.

    Args:
        response: Search response or log entry
        format_type: Output format (json, log, table, csv, summary)
        **kwargs: Additional format-specific options

    Returns:
        Formatted string

    """
    match format_type.lower():
        case "json":
            return format_json(response, **kwargs)
        case "log":
            return _format_as_log(response)
        case "table":
            return _format_as_table(response, **kwargs)
        case "csv":
            return _format_as_csv(response, **kwargs)
        case "summary":
            return _format_as_summary(response)
        case _:
            # Default to log format
            return _format_as_log(response)

format_summary

format_summary(response: SearchResponse) -> str

Format a summary of the search response.

Parameters:

Name	Type	Description	Default
response	SearchResponse	Search response	required

Returns:

Type	Description
str	Summary string

Source code in provide/foundation/integrations/openobserve/formatters.py

def format_summary(response: SearchResponse) -> str:
    """Format a summary of the search response.

    Args:
        response: Search response

    Returns:
        Summary string

    """
    lines = [
        f"Total hits: {response.total}",
        f"Returned: {len(response.hits)}",
        f"Query time: {response.took}ms",
        f"Scan size: {response.scan_size:,} bytes",
    ]

    if response.trace_id:
        lines.append(f"Trace ID: {response.trace_id}")

    if response.is_partial:
        lines.append("⚠️  Results are partial")

    if response.function_error:
        lines.append("Errors:")
        for error in response.function_error:
            lines.append(f"  - {error}")

    # Add level distribution if available
    level_counts: dict[str, int] = {}
    for hit in response.hits:
        level = hit.get("level", "UNKNOWN")
        level_counts[level] = level_counts.get(level, 0) + 1

    if level_counts:
        lines.append("\nLevel distribution:")
        for level, count in sorted(level_counts.items()):
            lines.append(f"  {level}: {count}")

    return "\n".join(lines)

format_table

format_table(
    response: SearchResponse,
    columns: list[str] | None = None,
) -> str

Format response as a table.

Parameters:

Name	Type	Description	Default
response	SearchResponse	Search response	required
columns	list[str] | None	Specific columns to include (None for all)	None

Returns:

Type	Description
str	Table string

Source code in provide/foundation/integrations/openobserve/formatters.py

def format_table(response: SearchResponse, columns: list[str] | None = None) -> str:
    """Format response as a table.

    Args:
        response: Search response
        columns: Specific columns to include (None for all)

    Returns:
        Table string

    """
    if not response.hits:
        return "No results found"

    # Determine columns if not provided
    if columns is None:
        columns = _determine_columns(response.hits)
        columns = _filter_internal_columns(columns)

    # Try to use tabulate if available
    try:
        return _format_with_tabulate(response.hits, columns)
    except ImportError:
        return _format_simple_table(response.hits, columns)