Logging

Simple logging utilities for Arshai framework with OpenTelemetry integration.

This module provides logging support similar to Taloan project - simple and clean, with optional OpenTelemetry integration when available.

Classes

class arshai.utils.logging.ApiEvent(direction, endpoint, message, status=None)

Bases: StructuredLogEvent

Event for API call logs

__init__(direction, endpoint, message, status=None)

set_duration(duration_ms)

Set the duration of the API call

log(logger_instance)

Log API event with appropriate level based on status

class arshai.utils.logging.ConnectionEvent(state, connection_type, entity_id, message)

Bases: StructuredLogEvent

Event for connection/disconnection logs

__init__(state, connection_type, entity_id, message)

class arshai.utils.logging.ErrorEvent(error_type, message, exception=None)

Bases: StructuredLogEvent

Event for error logs

__init__(error_type, message, exception=None)

log(logger_instance)

Log error event with error level

class arshai.utils.logging.HandlerInterface(*args, **kwargs)

Bases: Protocol

Protocol defining the handler interface

add_handler(sink, format, level, **kwargs)

remove_handler(handler_id)

__init__(*args, **kwargs)

class arshai.utils.logging.InterceptHandler(level=0)

Bases: Handler

Handler to intercept standard library logging into loguru

emit(record)

Do whatever it takes to actually log the specified logging record.

This version is intended to be implemented by subclasses and so raises a NotImplementedError.

class arshai.utils.logging.LogLevel(value)

Bases: str, Enum

Enum representing valid log levels

TRACE = 'TRACE'

DEBUG = 'DEBUG'

INFO = 'INFO'

SUCCESS = 'SUCCESS'

WARNING = 'WARNING'

ERROR = 'ERROR'

CRITICAL = 'CRITICAL'

classmethod validate(level)

Validate if a string is a valid log level

Return type:

bool

class arshai.utils.logging.LoggerAdapter(logger_instance=<loguru.logger handlers=[(id=1, level=20, sink=<stdout>)]>)

Bases: object

Adapter for loguru logger implementing LoggerInterface

__init__(logger_instance=<loguru.logger handlers=[(id=1, level=20, sink=<stdout>)]>)

debug(message, **kwargs)

info(message, **kwargs)

warning(message, **kwargs)

error(message, **kwargs)

critical(message, **kwargs)

bind(**kwargs)

class arshai.utils.logging.LoggerFactory(adapter=None, handler=None)

Bases: object

Factory for creating logger instances

__init__(adapter=None, handler=None)

create_logger(name, **context)

Create a simple logger with name

Return type:

LoggerInterface

create_correlation_logger(name, correlation_id=None, trace_id=None, span_id=None)

Create a logger with correlation ID and trace context

Return type:

LoggerInterface

create_trace_logger(name, current_span=None)

Create a logger with correlation ID extracted from the current OpenTelemetry span

Return type:

LoggerInterface

class arshai.utils.logging.LoggerInterface(*args, **kwargs)

Bases: Protocol

Protocol defining the logger interface

debug(message, **kwargs)

info(message, **kwargs)

warning(message, **kwargs)

error(message, **kwargs)

critical(message, **kwargs)

bind(**kwargs)

__init__(*args, **kwargs)

class arshai.utils.logging.LoggingConfig

Bases: object

Configuration container for logger settings

ENV_LOG_LEVEL = 'LOG_LEVEL'

ENV_LOG_FORMAT = 'LOG_FORMAT'

ENV_LOG_JSON = 'LOG_JSON'

DEFAULT_LOG_LEVEL = 'INFO'

DEFAULT_LOG_FORMAT = '<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | <cyan>{extra[name]}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>'

DEFAULT_LOG_JSON = False

classmethod get_log_level()

Get log level from environment or default

Return type:

str

classmethod get_log_format()

Get log format from environment or default

Return type:

str

classmethod is_json_logging()

Check if JSON logging is enabled

Return type:

bool

class arshai.utils.logging.LoggingFormat(value)

Bases: str, Enum

Enum representing logging formats

DEFAULT = '<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | <cyan>{extra[name]}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>'

JSON = '{message}'

CORRELATION = '<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | <cyan>{extra[name]}</cyan> | <yellow>trace_id={extra[trace_id]}</yellow> | <yellow>span_id={extra[span_id]}</yellow> | <yellow>correlation_id={extra[correlation_id]}</yellow> - <level>{message}</level>'

JSON_CORRELATION = '{message}'

class arshai.utils.logging.LoggingService(handler=None)

Bases: object

Central service for managing logging configuration

__init__(handler=None)

classmethod is_otel_enabled()

Check if OpenTelemetry is enabled

Return type:

bool

classmethod set_otel_enabled(enabled)

Set OpenTelemetry enabled status

configure(json_logging=None, level=None)

Configure the logging service

setup_standard_logging(level=None)

Configure standard library logging to use loguru

set_log_level(level)

Set global log level at runtime

class arshai.utils.logging.LoguruHandler(logger_instance=<loguru.logger handlers=[(id=1, level=20, sink=<stdout>)]>)

Bases: object

Handler for loguru logger implementing HandlerInterface

__init__(logger_instance=<loguru.logger handlers=[(id=1, level=20, sink=<stdout>)]>)

add_handler(sink=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, format=LoggingFormat.DEFAULT, level='INFO', **kwargs)

remove_handler(handler_id)

get_handlers()

class arshai.utils.logging.MetricEvent(metric_name, value, unit='')

Bases: StructuredLogEvent

Event for metric logs

__init__(metric_name, value, unit='')

class arshai.utils.logging.StructuredLogEvent(event_type, message)

Bases: object

Base class for structured log events

__init__(event_type, message)

add_data(**kwargs)

Add data to the event

get_data()

Get event data

log(logger_instance, level='info')

Log the event with the specified level

Functions

arshai.utils.logging.configure_logging(level=None, **kwargs)

Configure logging for the application (legacy compatibility).

Parameters:

• level (str) – Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)

• **kwargs – Other arguments (ignored for compatibility)

arshai.utils.logging.get_correlation_logger_from_trace(name, current_span=None)

Create a logger with correlation ID extracted from the current OpenTelemetry span. If no current span exists, creates a regular logger.

Parameters:

• name (str) – The logger name

• current_span – Optional current span to extract context from

Returns:

A loguru logger with trace context bound

arshai.utils.logging.get_logger(name)

Get a named logger instance with proper context

Parameters:

name (str) – The logger name, typically __name__ from calling module

Returns:

A loguru logger instance bound with context data

arshai.utils.logging.get_logger_with_correlation(name, correlation_id=None, trace_id=None, span_id=None)

Get a logger with correlation ID and trace context bound to the context. This is ideal for distributed tracing scenarios where logs need to be correlated across services.

Parameters:

• name (str) – The logger name

• correlation_id (Optional[str]) – Optional correlation ID for distributed tracing

• trace_id (Optional[str]) – Optional OpenTelemetry trace ID

• span_id (Optional[str]) – Optional OpenTelemetry span ID

Returns:

A loguru logger with context bound

arshai.utils.logging.get_simple_logger(name)

Get a simple loguru logger

arshai.utils.logging.get_trace_logger(name, current_span=None)

Get a logger with OpenTelemetry trace context

arshai.utils.logging.is_otel_enabled()

Check if OpenTelemetry is enabled

Return type:

bool

arshai.utils.logging.log_api(logger_instance, direction, endpoint, status=None, message='', extra_data=None, duration_ms=None)

Log API calls consistently

Parameters:

• logger_instance – The logger instance to use

• direction (str) – API direction (‘request’, ‘response’)

• endpoint (str) – API endpoint path

• status (Optional[int]) – Optional HTTP status code (for responses)

• message (str) – Human-readable message

• extra_data (Optional[Dict[str, Any]]) – Optional dictionary with additional context data

• duration_ms (Optional[float]) – Optional duration of the API call in milliseconds

arshai.utils.logging.log_connection(logger_instance, state, connection_type, entity_id, message, extra_data=None)

Log connection/disconnection events consistently

Parameters:

• logger_instance – The logger instance to use

• state (str) – Connection state (‘connected’, ‘disconnected’, etc.)

• connection_type (str) – Type of connection (‘sse’, ‘websocket’, ‘redis’, etc.)

• entity_id (str) – ID of the connected entity (session ID, user ID, etc.)

• message (str) – Human-readable message

• extra_data (Optional[Dict[str, Any]]) – Optional dictionary with additional context data

arshai.utils.logging.log_error(logger_instance, error_type, message, exception=None, extra_data=None)

Log errors consistently

Parameters:

• logger_instance – The logger instance to use

• error_type (str) – Category of error

• message (str) – Human-readable error message

• exception (Optional[Exception]) – Optional exception object

• extra_data (Optional[Dict[str, Any]]) – Optional dictionary with additional context data

arshai.utils.logging.log_event(logger_instance, level, event_type, message, data=None)

Log an event with consistent structure and optional context data

Parameters:

• logger_instance – The logger instance to use

• level (str) – The log level (debug, info, warning, error, critical)

• event_type (str) – Type of event (e.g., ‘connection’, ‘redis’, ‘api_call’)

• message (str) – Human-readable message

• data (Optional[Dict[str, Any]]) – Optional dictionary with additional context data

arshai.utils.logging.log_metric(logger_instance, metric_name, value, unit='', context=None)

Log metrics in a consistent format. This is complementary to OTEL metrics but useful for ad-hoc metrics that don’t need formal instrumentation.

Parameters:

• logger_instance – The logger instance to use

• metric_name (str) – Name of the metric

• value (Union[int, float]) – Numeric value of the metric

• unit (str) – Optional unit of the metric

• context (Optional[Dict[str, Any]]) – Optional context data for the metric

arshai.utils.logging.set_log_level(level)

Set the global logging level at runtime

Parameters:

level (Union[str, int]) – The log level to set (e.g., “DEBUG”, “INFO”, “WARNING”, “ERROR”)

arshai.utils.logging.setup_logging()

Configure loguru as the primary logging tool, intercept standard library logging