Commit 0a9531ec authored by Siarhei Khaletski (EPAM)'s avatar Siarhei Khaletski (EPAM) 🚩
Browse files

Added support of ADC ("Application Default Credentials") for GCP

parent 959891b7
Pipeline #67987 passed with stage
in 34 seconds
import aiohttp
import json
from typing import List, Optional
from typing import List, Optional, Type
import datetime
import os
import time
import jwt
from urllib.parse import quote_plus
......@@ -24,9 +25,9 @@ class GoogleAccountAuth(AuthBase):
:param scopes: scopes
with open(service_file, 'r') as f:
self.service_data = json.load(f)
self.service_file = service_file
self.service_data = self.get_service_data()
self.session = session
self.scopes = ' '.join(scopes or [])
self.access_token: Optional[str] = None
......@@ -38,6 +39,23 @@ class GoogleAccountAuth(AuthBase):
""" Return scheme """
return self._scheme
def get_service_data(self):
_service = self.service_file or os.environ.get('GOOGLE_APPLICATION_CREDENTIALS')
if not _service:
cloudsdk_config = os.environ.get('CLOUDSDK_CONFIG')
sdkpath = (cloudsdk_config
or os.path.join(os.path.expanduser('~'), '.config', 'gcloud'))
_service = os.path.join(sdkpath, 'application_default_credentials.json')
with open(_service, 'r') as f:
return json.load(f)
except TypeError:
return json.loads(
except Exception:
return None
async def header_value(self) -> str:
token = await self._get_token()
return f'{self._scheme} {token}'
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment