Move 'refresh' method to '_get_credentials_email' in base_google hook

airflow/providers/google/cloud/utils/credentials_provider.py

@@ -29,10 +29,8 @@
 import google.auth
 import google.auth.credentials
 import google.oauth2.service_account
-from google.auth import compute_engine, impersonated_credentials
+from google.auth import impersonated_credentials
 from google.auth.environment_vars import CREDENTIALS, LEGACY_PROJECT, PROJECT
-from google.auth.exceptions import RefreshError
-from google.auth.transport import _http_client
 
 from airflow.exceptions import AirflowException
 from airflow.providers.google.cloud._internal_client.secret_manager_client import _SecretManagerClient
@@ -253,16 +251,6 @@ def get_credentials_and_project(self) -> Tuple[google.auth.credentials.Credentia
 
             project_id = _get_project_id_from_service_account_email(self.target_principal)
 
-        if isinstance(credentials, compute_engine.Credentials):
-            try:
-                credentials.refresh(_http_client.Request())
-            except RefreshError as msg:
-                """
-                If the Compute Engine metadata service can't be reached in this case the instance has not
-                credentials.
-                """
-                self._log_debug(msg)
-
         return credentials, project_id
 
     def _get_credentials_using_keyfile_dict(self):

airflow/providers/google/common/hooks/base_google.py

@@ -34,8 +34,10 @@
 import tenacity
 from google.api_core.exceptions import Forbidden, ResourceExhausted, TooManyRequests
 from google.api_core.gapic_v1.client_info import ClientInfo
-from google.auth import _cloud_sdk
+from google.auth import _cloud_sdk, compute_engine
 from google.auth.environment_vars import CLOUD_SDK_CONFIG_DIR, CREDENTIALS
+from google.auth.exceptions import RefreshError
+from google.auth.transport import _http_client
 from googleapiclient import discovery
 from googleapiclient.errors import HttpError
 from googleapiclient.http import MediaIoBaseDownload, build_http, set_user_agent
@@ -274,6 +276,17 @@ def _get_credentials_email(self) -> str:
         If user authentication (e.g. gcloud auth) is used, it returns the e-mail account of that user.
         """
         credentials = self._get_credentials()
+
+        if isinstance(credentials, compute_engine.Credentials):
+            try:
+                credentials.refresh(_http_client.Request())
+            except RefreshError as msg:
+                """
+                If the Compute Engine metadata service can't be reached in this case the instance has not
+                credentials.
+                """
+                self.log.debug(msg)
+
         service_account_email = getattr(credentials, 'service_account_email', None)
         if service_account_email:
             return service_account_email