TokenManager

Manages authentication token lifecycle, including acquisition, renewal scheduling, and retry logic, to ensure continuous valid credentials for services.

class TokenManager:
    def __init__(
        self, identity_provider: IdentityProviderInterface, config: TokenManagerConfig
    ):
        self._idp = identity_provider
        self._config = config
        self._next_timer = None
        self._listener = None
        self._init_timer = None
        self._retries = 0

    def __del__(self):
        logger.info("Token manager are disposed")
        self.stop()

    def start(
        self,
        listener: CredentialsListener,
        skip_initial: bool = False,
    ) -> Callable[[], None]:
        self._listener = listener

        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            # Run loop in a separate thread to unblock main thread.
            loop = asyncio.new_event_loop()

            # Us

... (truncated -- full source via MCP)