logging_utils

logging_utils

Structured logging utilities.

Provides consistent logging configuration across the SDK using structlog.

configure_logging

configure_logging(level: str = 'INFO', json_format: bool = False, include_timestamp: bool = True) -> None

Configure structured logging for the SDK.

Parameters:

Name	Type	Description	Default
level	str	Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)	'INFO'
json_format	bool	Use JSON output format	False
include_timestamp	bool	Include timestamps in logs	True

Source code in ondine/utils/logging_utils.py

def configure_logging(
    level: str = "INFO",
    json_format: bool = False,
    include_timestamp: bool = True,
) -> None:
    """
    Configure structured logging for the SDK.

    Args:
        level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
        json_format: Use JSON output format
        include_timestamp: Include timestamps in logs
    """
    global _logging_configured

    # Set stdlib logging level
    logging.basicConfig(
        format="%(message)s",
        stream=sys.stdout,
        level=getattr(logging, level.upper()),
    )

    # Configure structlog processors
    processors = [
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.StackInfoRenderer(),
    ]

    if include_timestamp:
        processors.append(structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S"))

    if json_format:
        processors.append(structlog.processors.JSONRenderer())
    else:
        # Use custom compact console renderer (no padding)
        processors.append(_compact_console_renderer)

    structlog.configure(
        processors=processors,
        wrapper_class=structlog.make_filtering_bound_logger(
            getattr(logging, level.upper())
        ),
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory(),
        cache_logger_on_first_use=True,
    )

    _logging_configured = True

get_logger

get_logger(name: str) -> structlog.BoundLogger

Get a structured logger instance.

Auto-configures logging on first use if not already configured.

Parameters:

Name	Type	Description	Default
name	str	Logger name (typically name)	required

Returns:

Type	Description
BoundLogger	Configured structlog logger

Source code in ondine/utils/logging_utils.py

def get_logger(name: str) -> structlog.BoundLogger:
    """
    Get a structured logger instance.

    Auto-configures logging on first use if not already configured.

    Args:
        name: Logger name (typically __name__)

    Returns:
        Configured structlog logger
    """
    global _logging_configured

    # Auto-configure logging on first use
    if not _logging_configured:
        configure_logging()

    return structlog.get_logger(name)

sanitize_for_logging

sanitize_for_logging(data: dict[str, Any]) -> dict[str, Any]

Sanitize sensitive data for logging.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Dictionary potentially containing sensitive data	required

Returns:

Type	Description
dict[str, Any]	Sanitized dictionary

Source code in ondine/utils/logging_utils.py

def sanitize_for_logging(data: dict[str, Any]) -> dict[str, Any]:
    """
    Sanitize sensitive data for logging.

    Args:
        data: Dictionary potentially containing sensitive data

    Returns:
        Sanitized dictionary
    """
    sensitive_keys = {
        "api_key",
        "password",
        "secret",
        "token",
        "authorization",
        "credential",
    }

    sanitized = {}
    for key, value in data.items():
        key_lower = key.lower()
        if any(sensitive in key_lower for sensitive in sensitive_keys):
            sanitized[key] = "***REDACTED***"
        elif isinstance(value, dict):
            sanitized[key] = sanitize_for_logging(value)
        else:
            sanitized[key] = value

    return sanitized