def get_token()

in azure-kusto-data/azure/kusto/data/_token_providers.py [0:0]

• 12 lines of code
• 3 McCabe index (conditional complexity)

    def get_token(self):
        """Get a token silently from cache or authenticate if cached token is not found"""

        @distributed_trace(name_of_span=f"{self.name()}.get_token", tracing_attributes=self.context(), kind=SpanKind.CLIENT)
        def _get_token():
            if self.is_async:
                raise KustoAsyncUsageError("get_token", self.is_async)
            self._init_once()

            token = self._get_token_from_cache_impl()
            if token is None:
                with self._lock:
                    token = MonitoredActivity.invoke(self._get_token_impl, name_of_span=f"{self.name()}.get_token_impl", tracing_attributes=self.context())
            return self._valid_token_or_throw(token)

        return _get_token()