def __init__()

in aws_advanced_python_wrapper/utils/telemetry/default_telemetry_factory.py [0:0]

• 35 lines of code
• 9 McCabe index (conditional complexity)

    def __init__(self, properties: Properties):
        self._enable_telemetry = WrapperProperties.ENABLE_TELEMETRY.get(properties)
        self._telemetry_submit_toplevel = WrapperProperties.TELEMETRY_SUBMIT_TOPLEVEL.get(properties)
        self._telemetry_traces_backend = WrapperProperties.TELEMETRY_TRACES_BACKEND.get(properties)
        self._telemetry_metrics_backend = WrapperProperties.TELEMETRY_METRICS_BACKEND.get(properties)

        self._traces_telemetry_factory: TelemetryFactory
        self._metrics_telemetry_factory: TelemetryFactory

        if self._enable_telemetry:
            if self._telemetry_traces_backend is not None:
                traces_backend = self._telemetry_traces_backend.upper()
                if traces_backend == "OTLP":
                    self._traces_telemetry_factory = OpenTelemetryFactory()
                elif traces_backend == "XRAY":
                    self._traces_telemetry_factory = XRayTelemetryFactory()
                elif traces_backend == "NONE":
                    self._traces_telemetry_factory = NullTelemetryFactory()
                else:
                    raise RuntimeError(Messages.get_formatted(
                        "DefaultTelemetryFactory.InvalidTracingBackend", self._telemetry_traces_backend))
            else:
                raise RuntimeError(Messages.get_formatted("DefaultTelemetryFactory.NoTracingBackendProvided"))

            if self._telemetry_metrics_backend is not None:
                metrics_backend = self._telemetry_metrics_backend.upper()
                if metrics_backend == "OTLP":
                    self._metrics_telemetry_factory = OpenTelemetryFactory()
                elif metrics_backend == "NONE":
                    self._metrics_telemetry_factory = NullTelemetryFactory()
                else:
                    raise RuntimeError(Messages.get_formatted(
                        "DefaultTelemetryFactory.InvalidMetricsBackend", self._telemetry_metrics_backend))
            else:
                raise RuntimeError(Messages.get_formatted("DefaultTelemetryFactory.NoMetricsBackendProvided"))

        else:
            self._traces_telemetry_factory = NullTelemetryFactory()
            self._metrics_telemetry_factory = NullTelemetryFactory()