telemetry_utils.py

#

Utility functions for OpenTelemetry instrumentation with graceful degradation.

This module provides a safe way to import and use OpenTelemetry decorators and metrics that gracefully degrades when OpenTelemetry packages are not installed.

#

Define no-op implementations first (always available for testing) No-op decorator when OpenTelemetry is not available

def _no_op_trace_decorator(operation_name):

#

    def decorator(func):
        return func

    return decorator

#

No-op metrics class when OpenTelemetry is not available

class NoOpMetrics:

#

    def record_avatar_generated(self, *args, **kwargs):
        pass

#

    def record_avatar_request(self, *args, **kwargs):
        pass

#

    def record_cache_hit(self, *args, **kwargs):
        pass

#

    def record_cache_miss(self, *args, **kwargs):
        pass

#

    def record_external_request(self, *args, **kwargs):
        pass

#

    def record_file_upload(self, *args, **kwargs):
        pass

#

Safe import pattern for OpenTelemetry

try:
    from .opentelemetry_middleware import (
        trace_avatar_operation,
        trace_file_upload,
        trace_authentication,
        get_avatar_metrics,
    )

#

Get the actual metrics instance

    avatar_metrics = get_avatar_metrics()

#

OpenTelemetry is available

    TELEMETRY_AVAILABLE = True

except ImportError:

#

OpenTelemetry packages not installed - use no-op implementations

    trace_avatar_operation = _no_op_trace_decorator
    trace_file_upload = _no_op_trace_decorator
    trace_authentication = _no_op_trace_decorator

    avatar_metrics = NoOpMetrics()

#

OpenTelemetry is not available

    TELEMETRY_AVAILABLE = False

#

Get all telemetry decorators in a single call.

Returns: tuple: (trace_avatar_operation, trace_file_upload, trace_authentication)

def get_telemetry_decorators():

#

    return trace_avatar_operation, trace_file_upload, trace_authentication

#

Get the telemetry metrics instance.

Returns: AvatarMetrics or NoOpMetrics: The metrics instance

def get_telemetry_metrics():

#

    return avatar_metrics

#

Check if OpenTelemetry is available and working.

Returns: bool: True if OpenTelemetry is available, False otherwise

def is_telemetry_available():

#

Check the actual metrics instance type rather than relying on the module-level variable which can be affected by test mocking

    return not isinstance(avatar_metrics, NoOpMetrics)