token_manager¶

class TokenManager:
    """Caches and transparently refreshes a Graph API app-only access token
    for a single Azure AD app registration."""

    def __init__(
        self,
        tenant_id: str,
        client_id: str,
        client_secret: str,
        scope: str = "https://graph.microsoft.com/.default",
    ) -> None:
        """
        Args:
            tenant_id (str): Azure AD tenant ID.
            client_id (str): App registration's client ID.
            client_secret (str): App registration's client secret.
            scope (str): OAuth2 scope to request the token for.
        """
        self.token_url = (
            f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
        )
        self.client_id = client_id
        self.client_secret = client_secret
        self.scope = scope
        self.access_token: str | None = None
        self.expires_at: float = 0

    def get_token(self) -> str:
        """
        Returns a valid access token, refreshing it first if needed.

        Returns:
            str: A valid bearer access token.

        Raises:
            TokenNotSetError: If a refresh completed without setting a token.
        """
        # return a valid token and if the token has expired or is about to expire , request a new token.
        if (
            self.access_token is None or time.time() >= self.expires_at - 300
        ):  # Refresh if less than 5 minutes remain
            self._refresh_token()
        if self.access_token is None:
            raise TokenNotSetError
        return self.access_token

    def _refresh_token(self) -> None:
        """
        Requests a fresh access token from Azure AD and stores it.

        Raises:
            TokenRequestError: If the token request fails.
        """
        # request a new token for AZURE AD
        token_data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": self.scope,
        }
        response = requests.post(self.token_url, data=token_data, timeout=10)
        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
            raise TokenRequestError(e, response.text) from e

        token_data = response.json()
        self.access_token = token_data["access_token"]
        self.expires_at = time.time() + token_data.get("expires_in", 3600)