h2o-authn

import os import httpx import asyncio from h2o_authn import ClientCredentialsTokenProvider, AuthCache async def authenticate_and_get_token(): auth_url = os.environ.get("H2O_AUTHN_URI", "") client_id = os.environ.get("H2O_AUTHN_CLIENT_ID", "") client_secret = os.environ.get("H2O_AUTHN_CLIENT_SECRET", "") if not all([auth_url, client_id, client_secret]): print("Please set H2O_AUTHN_URI, H2O_AUTHN_CLIENT_ID, and H2O_AUTHN_CLIENT_SECRET environment variables.") return async with httpx.AsyncClient() as client: token_provider = ClientCredentialsTokenProvider( auth_url=auth_url, client_id=client_id, client_secret=client_secret, client=client, ) # Optional: Wrap with AuthCache for token caching and refresh cached_token_provider = AuthCache(token_provider) try: token = await cached_token_provider.get_token() print(f"Successfully obtained token. Access Token (first 10 chars): {token.access_token[:10]}...") print(f"Token expires in: {token.expires_in} seconds") except Exception as e: print(f"Authentication failed: {e}") if __name__ == "__main__": asyncio.run(authenticate_and_get_token())