def _embeddings_wrapper()

in instrumentation/elastic-opentelemetry-instrumentation-openai/src/opentelemetry/instrumentation/openai/__init__.py [0:0]

• 28 lines of code
• 4 McCabe index (conditional complexity)

    def _embeddings_wrapper(self, wrapped, instance, args, kwargs):
        span_attributes = _get_embeddings_attributes_from_wrapper(instance, kwargs)

        span_name = _span_name_from_attributes(span_attributes)
        with self.tracer.start_as_current_span(
            name=span_name,
            kind=SpanKind.CLIENT,
            attributes=span_attributes,
            # this is important to avoid having the span closed before ending the stream
            end_on_exit=False,
        ) as span:
            start_time = default_timer()
            try:
                result = wrapped(*args, **kwargs)
            except Exception as exc:
                span.set_status(StatusCode.ERROR, str(exc))
                span.set_attribute(ERROR_TYPE, exc.__class__.__qualname__)
                span.end()
                error_attributes = {**span_attributes, ERROR_TYPE: exc.__class__.__qualname__}
                _record_operation_duration_metric(self.operation_duration_metric, error_attributes, start_time)
                raise

            response_attributes = _get_embeddings_attributes_from_response(result.model, result.usage)
            if span.is_recording():
                for k, v in response_attributes.items():
                    span.set_attribute(k, v)

            metrics_attributes = {**span_attributes, **response_attributes}
            _record_token_usage_metrics(self.token_usage_metric, metrics_attributes, result.usage)
            _record_operation_duration_metric(self.operation_duration_metric, metrics_attributes, start_time)

            span.end()

            return result