Function _get_management_api_token

api/auth0_manager.py:22–57

Fetches a fresh access token for the Auth0 Management API. Uses in-memory cache with 4-hour expiration to reduce API calls.

()

_token_expires_at = 0  # epoch timestamp when token expires

def _get_management_api_token() -> str:
    """
    Fetches a fresh access token for the Auth0 Management API.
    Uses in-memory cache with 4-hour expiration to reduce API calls.
    """
    global _mgmt_token, _token_expires_at

    # Check if cached token is still valid
    if _mgmt_token and time.time() < _token_expires_at:
        return _mgmt_token

    if not all([AUTH0_DOMAIN, MGMT_API_CLIENT_ID, MGMT_API_CLIENT_SECRET]):
        logger.error("Auth0 Management API credentials are not fully configured.")
        raise ValueError("Auth0 Management API credentials are not set in environment.")

    payload = {
        "client_id": MGMT_API_CLIENT_ID,
        "client_secret": MGMT_API_CLIENT_SECRET,
        "audience": MGMT_API_AUDIENCE,
        "grant_type": "client_credentials"
    }
    headers = {'content-type': "application/json"}

    try:
        response = requests.post(f"https://{AUTH0_DOMAIN}/oauth/token", json=payload, headers=headers)
        response.raise_for_status() # Raise an exception for HTTP errors (4xx or 5xx)

        # Cache the token for 4 hours
        _mgmt_token = response.json()["access_token"]
        _token_expires_at = time.time() + (4 * 60 * 60)  # 4 hours in seconds

        logger.info("Successfully fetched Auth0 Management API token.")
        return _mgmt_token
    except requests.exceptions.RequestException as e:
        logger.error(f"Failed to get Auth0 Management API token: {e}")
        raise

# Sentinel value to indicate a field should be cleared (deleted) from metadata
CLEAR_FIELD = "__CLEAR__"