Spaces:
Running
Running
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 | |