Method get_token

src/openai/auth/_workload.py:190–217

(self)

        self._condition = threading.Condition(self._lock)

    def get_token(self) -> str:
        with self._lock:
            while self._refreshing and self._token_unusable():
                self._condition.wait()

            if not self._token_unusable() and not self._needs_refresh():
                return cast(str, self._cached_token)

            if self._refreshing:
                while self._refreshing:
                    self._condition.wait()
                token = self._cached_token  # type: ignore[unreachable]
                if self._token_unusable():
                    raise RuntimeError("Token is unusable after refresh completed")
                return cast(str, token)

            self._refreshing = True

        try:
            self._perform_refresh()
            with self._lock:
                if self._token_unusable():
                    raise RuntimeError("Token is unusable after refresh completed")
                return cast(str, self._cached_token)
        finally:
            with self._lock:
                self._refreshing = False
                self._condition.notify_all()

    async def get_token_async(self) -> str:
        return await to_thread(self.get_token)