def __init__()

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

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

    def __init__(self, tracer: Tracer, name: str, trace_level: TelemetryTraceLevel,
                 start_time: Optional[int] = None, link_span: Optional[Span] = None):
        self._name = name
        self._tracer = tracer
        self._span: Optional[Span]
        self._meter: Meter
        self._token: Optional[object] = None

        current_span: Span = trace.get_current_span()  # type: ignore
        is_root = (current_span is None or current_span == trace.INVALID_SPAN)

        if is_root and trace_level == TelemetryTraceLevel.NESTED:
            trace_level = TelemetryTraceLevel.TOP_LEVEL

        links: Sequence[trace_api.Link] = ()
        if trace_level in [TelemetryTraceLevel.FORCE_TOP_LEVEL, TelemetryTraceLevel.TOP_LEVEL]:
            if link_span is not None:
                links = [trace.Link(link_span.get_span_context())]
            else:
                if not is_root:
                    links = [trace.Link(current_span.get_span_context())]

            self._span = self._tracer.start_span(self._name, context=context_api.Context(),
                                                 links=links, start_time=start_time)  # type: ignore
            if not is_root:
                self.set_attribute(TelemetryConst.TRACE_NAME_ANNOTATION, self._name)

            ctx = trace.set_span_in_context(self._span)  # type: ignore
            self._token = context_api.attach(ctx)
            logger.debug("OpenTelemetryContext.TelemetryTraceID", self._name,
                         self._span.get_span_context().trace_id)  # type: ignore

        elif trace_level == TelemetryTraceLevel.NESTED:
            if link_span is not None:
                links = [trace.Link(link_span.get_span_context())]

            self._span = self._tracer.start_span(self._name, links=links, start_time=start_time)  # type: ignore
            ctx = trace.set_span_in_context(self._span)  # type: ignore
            self._token = context_api.attach(ctx)
            self.set_attribute(TelemetryConst.TRACE_NAME_ANNOTATION, self._name)

        elif trace_level == TelemetryTraceLevel.NO_TRACE:
            self._span = None