| import json |
| import os |
| import logging |
| from typing import Optional |
|
|
| import google.auth |
| import google.auth.credentials |
| from google.oauth2 import service_account |
|
|
| logger = logging.getLogger(__name__) |
|
|
| _SCOPES = ["https://www.googleapis.com/auth/cloud-platform"] |
|
|
|
|
| def get_google_credentials() -> google.auth.credentials.Credentials: |
| """ |
| Resolution order: |
| 1. GOOGLE_CREDENTIALS_JSON env var (raw JSON string) β service account info |
| 2. GOOGLE_APPLICATION_CREDENTIALS env var set & file exists β service account file |
| 3. Fallback β ADC (gcloud auth application-default login) |
| """ |
| creds_json: Optional[str] = os.environ.get("GOOGLE_CREDENTIALS_JSON") |
| if creds_json: |
| logger.info("Google auth: service account dari GOOGLE_CREDENTIALS_JSON") |
| |
| try: |
| import base64 |
| decoded = base64.b64decode(creds_json).decode("utf-8") |
| info = json.loads(decoded) |
| except Exception: |
| info = json.loads(creds_json) |
| return service_account.Credentials.from_service_account_info(info, scopes=_SCOPES) |
|
|
| creds_path: Optional[str] = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS") |
| if creds_path: |
| if not os.path.isfile(creds_path): |
| raise FileNotFoundError( |
| f"GOOGLE_APPLICATION_CREDENTIALS='{creds_path}' tapi file tidak ditemukan." |
| ) |
| logger.info("Google auth: service account dari %s", creds_path) |
| return service_account.Credentials.from_service_account_file(creds_path, scopes=_SCOPES) |
|
|
| logger.info("Google auth: menggunakan ADC (gcloud login)") |
| credentials, _ = google.auth.default(scopes=_SCOPES) |
| return credentials |
|
|