Spaces:
Sleeping
Sleeping
| from google.oauth2 import service_account | |
| from google.auth.transport.requests import Request | |
| from dotenv import load_dotenv | |
| import json | |
| import os | |
| import logging | |
| def get_access_token(): | |
| # Load service account credentials from JSON file or environment variable | |
| credentials = service_account.Credentials.from_service_account_info( | |
| json.loads(os.getenv('ACCOUNT_CREDS')), | |
| scopes=['https://www.googleapis.com/auth/cloud-platform'] | |
| ) | |
| # Refresh token if needed | |
| if credentials.expired: | |
| credentials.refresh(Request()) | |
| return credentials.token | |
| def get_access_token_v1(): | |
| """ | |
| Generate a Google Cloud access token from service account credentials. | |
| Returns the access token string or raises an exception if failed. | |
| """ | |
| try: | |
| # Set up logging | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| # Get credentials from environment variable | |
| creds_json = os.getenv('ACCOUNT_CREDS') | |
| if not creds_json: | |
| raise ValueError("ACCOUNT_CREDS environment variable not found") | |
| # Parse credentials JSON | |
| try: | |
| creds_dict = json.loads(creds_json) | |
| except json.JSONDecodeError: | |
| raise ValueError("Invalid JSON in ACCOUNT_CREDS") | |
| # Create credentials object | |
| credentials = service_account.Credentials.from_service_account_info( | |
| creds_dict, | |
| scopes=['https://www.googleapis.com/auth/cloud-platform'] | |
| ) | |
| # Ensure token is valid and refresh if needed | |
| if not credentials.valid: | |
| logger.info("Token expired or invalid, refreshing...") | |
| credentials.refresh(Request()) | |
| if not credentials.token: | |
| raise ValueError("No token generated after refresh") | |
| logger.info("Successfully generated access token") | |
| return credentials.token | |
| except Exception as e: | |
| logger.error(f"Error generating access token: {str(e)}") | |
| raise | |