content: #  Copyright (c) ZenML GmbH 2022. All Rights Reserved.
  #
  #  Licensed under the Apache License, Version 2.0 (the "License");
  #  you may not use this file except in compliance with the License.
  #  You may obtain a copy of the License at:
  #
  #       https://www.apache.org/licenses/LICENSE-2.0
  #
  #  Unless required by applicable law or agreed to in writing, software
  #  distributed under the License is distributed on an "AS IS" BASIS,
  #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  #  or implied. See the License for the specific language governing
  #  permissions and limitations under the License.
  """Implementation of the Google credentials mixin."""
  
  from typing import TYPE_CHECKING, Optional, Tuple, cast
  
  from pydantic import Field
  
  from zenml.logger import get_logger
  from zenml.stack.stack_component import StackComponent, StackComponentConfig
  
  if TYPE_CHECKING:
      from google.auth.credentials import Credentials
  
  
  logger = get_logger(__name__)
  
  
  class GoogleCredentialsConfigMixin(StackComponentConfig):
      """Config mixin for Google Cloud Platform credentials.
  
      Provides common GCP authentication configuration for stack components.
      Field descriptions are defined inline using Field() descriptors.
      """
  
      project: Optional[str] = Field(
          default=None,
          description="Google Cloud Project ID. Auto-detected from environment if not specified.",
      )
      service_account_path: Optional[str] = Field(
          default=None,
          description="Path to service account JSON key file for authentication. "
          "Uses Application Default Credentials if not provided.",
      )
  
  
  class GoogleCredentialsMixin(StackComponent):
      """StackComponent mixin to get Google Cloud Platform credentials."""
  
      _gcp_credentials: Optional["Credentials"] = None
      _gcp_project_id: Optional[str] = None
  
      @property
      def config(self) -> GoogleCredentialsConfigMixin:
          """Returns the `GoogleCredentialsConfigMixin` config.
  
          Returns:
              The configuration.
          """
          return cast(GoogleCredentialsConfigMixin, self._config)
  
      @property
      def gcp_project_id(self) -> str:
          """Get the GCP project ID.
  
          Returns:
              The GCP project ID.
          """
          if self._gcp_project_id is None:
              _, self._gcp_project_id = self._get_authentication()
  
          return self._gcp_project_id
  
      def _get_authentication(self) -> Tuple["Credentials", str]:
          """Get GCP credentials and the project ID associated with the credentials.
  
          If `service_account_path` is provided, then the credentials will be
          loaded from the file at that path. Otherwise, the default credentials
          will be used.
  
          Returns:
              A tuple containing the credentials and the project ID associated to
              the credentials.
  
          Raises:
              RuntimeError: If the linked connector returns an unexpected type of
                  credentials.
          """
          from google.auth import default, load_credentials_from_file
          from google.auth.credentials import Credentials
  
          from zenml.integrations.gcp.service_connectors import (
              GCPServiceConnector,
          )
  
          if self.connector_has_expired():
              self._gcp_credentials = None
  
          if self._gcp_credentials and self._gcp_project_id:
              return self._gcp_credentials, self._gcp_project_id
  
          connector = self.get_connector()
          if connector:
              credentials = connector.connect()
              if not isinstance(credentials, Credentials) or not isinstance(
                  connector, GCPServiceConnector
              ):
                  raise RuntimeError(
                      f"Expected google.auth.credentials.Credentials while "
                      "trying to use the linked connector, but got "
                      f"{type(credentials)}."
                  )
              self._gcp_credentials = credentials
              self._gcp_project_id = connector.config.gcp_project_id
              return credentials, connector.config.gcp_project_id
  
          if self.config.service_account_path:
              credentials, project_id = load_credentials_from_file(
                  self.config.service_account_path
              )
          else:
              credentials, project_id = default()
  
          if self.config.project and self.config.project != project_id:
              logger.warning(
                  "Authenticated with project `%s`, but this %s is "
                  "configured to use the project `%s`.",
                  project_id,
                  self.type,
                  self.config.project,
              )
  
          # If the project was set in the configuration, use it. Otherwise, use
          # the project that was used to authenticate.
          project_id = self.config.project if self.config.project else project_id
          self._gcp_credentials = credentials
          self._gcp_project_id = project_id
          return credentials, project_id
  