Add Cloud Run Operator

airflow/providers/google/cloud/hooks/cloud_run.py

@@ -0,0 +1,60 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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
+#
+#   http://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.
+"""This module contains a Google Cloud Run Hook."""
+from __future__ import annotations
+
+from typing import Sequence
+
+import google.auth.transport.requests
+
+from airflow.hooks.base import BaseHook
+from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
+
+
+class CloudRunHook(GoogleBaseHook):
+    """Hook for Google Cloud Run."""
+
+    def __init__(
+        self,
+        gcp_conn_id: str = "google_cloud_default",
+        cloud_run_conn_id: str = "http_default",
+        delegate_to: str | None = None,
+        impersonation_chain: str | Sequence[str] | None = None,
+    ) -> None:
+        super().__init__(
+            gcp_conn_id=gcp_conn_id, delegate_to=delegate_to, impersonation_chain=impersonation_chain
+        )
+        self.cloud_run_conn_id = cloud_run_conn_id
+
+    def get_conn(self) -> dict:
+        """
+        Retrieves HTTP authentication header allowing
+        authenticated Google Cloud Run call.
+        :return: Authentication header
+        :rtype: dict
+        """
+        http_connection = BaseHook.get_connection(self.cloud_run_conn_id)
+        credentials = self.get_id_token_credentials(target_audience=http_connection.host)
+        auth_req = google.auth.transport.requests.Request()
+        credentials.refresh(auth_req)
+
+        authentication_header = {
+            "Authorization": f"Bearer {credentials.token}",
+            "Content-Type": "application/json",
+        }
+        return authentication_header

airflow/providers/google/cloud/operators/cloud_run.py

@@ -0,0 +1,64 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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
+#
+#   http://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.
+"""This module contains a Google CloudRun operator."""
+from __future__ import annotations
+
+from typing import Sequence
+
+from airflow.providers.google.cloud.hooks.cloud_run import CloudRunHook
+from airflow.providers.http.operators.http import SimpleHttpOperator
+
+
+class CloudRunOperator(SimpleHttpOperator):
+    """
+    Performs an authenticated call against CloudRun
+
+    Under the hood, this operator use SimpleHttpOperator.
+    Check the documentation of SimpleHttpOperator for extra options.
+
+    This operator use the GCP connection to get a token and add it to http request Header.
+
+    .. seealso::
+        For more information on how to use this operator, take a look at the guide:
+        :ref:`howto/operator:CloudRunOperator`
+
+    :param http_conn_id: The :ref:`CloudRun http connection<howto/connection:http>` to run
+        the operator against
+    :param gcp_conn_id: (Optional) The connection ID used to connect to Google Cloud.
+    """
+
+    template_fields: Sequence[str] = ("gcp_conn_id",)
+
+    def __init__(
+        self,
+        *,
+        gcp_conn_id: str = "google_cloud_default",
+        **kwargs,
+    ) -> None:
+        super().__init__(**kwargs)
+        self.gcp_conn_id = gcp_conn_id
+
+    def execute(self, context):
+        cloud_run = CloudRunHook(
+            gcp_conn_id=self.gcp_conn_id,
+            cloud_run_conn_id=self.http_conn_id,
+        )
+
+        authentication_header = cloud_run.get_conn()
+        self.headers = {**authentication_header, **self.headers}
+        return super().execute(context)

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

@@ -33,6 +33,7 @@
 import google.oauth2.service_account
 from google.auth import impersonated_credentials
 from google.auth.environment_vars import CREDENTIALS, LEGACY_PROJECT, PROJECT
+from google.oauth2 import service_account
 
 from airflow.exceptions import AirflowException
 from airflow.providers.google.cloud._internal_client.secret_manager_client import _SecretManagerClient
@@ -199,10 +200,12 @@ def __init__(
         key_secret_name: str | None = None,
         key_secret_project_id: str | None = None,
         scopes: Collection[str] | None = None,
+        target_audience: str | None = None,
         delegate_to: str | None = None,
         disable_logging: bool = False,
         target_principal: str | None = None,
         delegates: Sequence[str] | None = None,
+        is_id_token_credentials: bool = False,
     ) -> None:
         super().__init__()
         key_options = [key_path, key_secret_name, keyfile_dict]
@@ -216,10 +219,12 @@ def __init__(
         self.key_secret_name = key_secret_name
         self.key_secret_project_id = key_secret_project_id
         self.scopes = scopes
+        self.target_audience = target_audience
         self.delegate_to = delegate_to
         self.disable_logging = disable_logging
         self.target_principal = target_principal
         self.delegates = delegates
+        self.is_id_token_credentials = is_id_token_credentials
 
     def get_credentials_and_project(self) -> tuple[google.auth.credentials.Credentials, str]:
         """
@@ -263,10 +268,7 @@ def _get_credentials_using_keyfile_dict(self):
         # Depending on how the JSON was formatted, it may contain
         # escaped newlines. Convert those to actual newlines.
         self.keyfile_dict["private_key"] = self.keyfile_dict["private_key"].replace("\\n", "\n")
-        credentials = google.oauth2.service_account.Credentials.from_service_account_info(
-            self.keyfile_dict, scopes=self.scopes
-        )
-        project_id = credentials.project_id
+        credentials, project_id = self._get_credentials_using_info(self.keyfile_dict)
         return credentials, project_id
 
     def _get_credentials_using_key_path(self):
@@ -277,10 +279,7 @@ def _get_credentials_using_key_path(self):
             raise AirflowException("Unrecognised extension for key file.")
 
         self._log_debug("Getting connection using JSON key file %s", self.key_path)
-        credentials = google.oauth2.service_account.Credentials.from_service_account_file(
-            self.key_path, scopes=self.scopes
-        )
-        project_id = credentials.project_id
+        credentials, project_id = self._get_credentials_using_file(self.key_path)
         return credentials, project_id
 
     def _get_credentials_using_key_secret_name(self):
@@ -305,10 +304,7 @@ def _get_credentials_using_key_secret_name(self):
         except json.decoder.JSONDecodeError:
             raise AirflowException("Key data read from GCP Secret Manager is not valid JSON.")
 
-        credentials = google.oauth2.service_account.Credentials.from_service_account_info(
-            keyfile_dict, scopes=self.scopes
-        )
-        project_id = credentials.project_id
+        credentials, project_id = self._get_credentials_using_info(keyfile_dict)
         return credentials, project_id
 
     def _get_credentials_using_adc(self):
@@ -318,6 +314,34 @@ def _get_credentials_using_adc(self):
         credentials, project_id = google.auth.default(scopes=self.scopes)
         return credentials, project_id
 
+    def _get_credentials_using_file(self, file):
+        if self.is_id_token_credentials:
+            credentials = service_account.IDTokenCredentials.from_service_account_file(
+                file,
+                target_audience=self.target_audience,
+            )
+            project_id = None
+        else:
+            credentials = google.oauth2.service_account.Credentials.from_service_account_file(
+                file, scopes=self.scopes
+            )
+            project_id = credentials.project_id
+        return credentials, project_id
+
+    def _get_credentials_using_info(self, info):
+        if self.is_id_token_credentials:
+            credentials = service_account.IDTokenCredentials.from_service_account_info(
+                info,
+                target_audience=self.target_audience,
+            )
+            project_id = None
+        else:
+            credentials = google.oauth2.service_account.Credentials.from_service_account_info(
+                info, scopes=self.scopes
+            )
+            project_id = credentials.project_id
+        return credentials, project_id
+
     def _log_info(self, *args, **kwargs) -> None:
         if not self.disable_logging:
             self.log.info(*args, **kwargs)
@@ -332,6 +356,16 @@ def get_credentials_and_project_id(*args, **kwargs) -> tuple[google.auth.credent
     return _CredentialProvider(*args, **kwargs).get_credentials_and_project()
 
 
+def get_id_token_credentials(*args, **kwargs) -> tuple[google.auth.credentials.Credentials]:
+    """Returns the ID Token type Credentials object for Google API."""
+    kwargs = {**kwargs, "is_id_token_credentials": True}
+    id_token_credentials, _ = _CredentialProvider(
+        *args,
+        **kwargs,
+    ).get_credentials_and_project()
+    return id_token_credentials
+
+
 def _get_scopes(scopes: str | None = None) -> Sequence[str]:
     """
     Parse a comma-separated string containing OAuth2 scopes if `scopes` is provided.

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

@@ -52,6 +52,7 @@
     _get_scopes,
     _get_target_principal_and_delegates,
     get_credentials_and_project_id,
+    get_id_token_credentials,
 )
 from airflow.providers.google.common.consts import CLIENT_INFO
 from airflow.utils.process_utils import patch_environ
@@ -235,11 +236,7 @@ def __init__(
         self._cached_credentials: google.auth.credentials.Credentials | None = None
         self._cached_project_id: str | None = None
 
-    def get_credentials_and_project_id(self) -> tuple[google.auth.credentials.Credentials, str | None]:
-        """Returns the Credentials object for Google API and the associated project_id"""
-        if self._cached_credentials is not None:
-            return self._cached_credentials, self._cached_project_id
-
+    def get_connection_info(self):
         key_path: str | None = self._get_field("key_path", None)
         try:
             keyfile_dict: str | dict[str, str] | None = self._get_field("keyfile_dict", None)
@@ -253,14 +250,24 @@ def get_credentials_and_project_id(self) -> tuple[google.auth.credentials.Creden
             raise AirflowException("Invalid key JSON.")
         key_secret_name: str | None = self._get_field("key_secret_name", None)
         key_secret_project_id: str | None = self._get_field("key_secret_project_id", None)
+        return {
+            "key_path": key_path,
+            "keyfile_dict": keyfile_dict_json,
+            "key_secret_name": key_secret_name,
+            "key_secret_project_id": key_secret_project_id,
+        }
+
+    def get_credentials_and_project_id(self) -> tuple[google.auth.credentials.Credentials, str | None]:
+        """Returns the Credentials object for Google API and the associated project_id"""
+        if self._cached_credentials is not None:
+            return self._cached_credentials, self._cached_project_id
+
+        connection_info = self.get_connection_info()
 
         target_principal, delegates = _get_target_principal_and_delegates(self.impersonation_chain)
 
         credentials, project_id = get_credentials_and_project_id(
-            key_path=key_path,
-            keyfile_dict=keyfile_dict_json,
-            key_secret_name=key_secret_name,
-            key_secret_project_id=key_secret_project_id,
+            **connection_info,
             scopes=self.scopes,
             delegate_to=self.delegate_to,
             target_principal=target_principal,
@@ -276,6 +283,15 @@ def get_credentials_and_project_id(self) -> tuple[google.auth.credentials.Creden
 
         return credentials, project_id
 
+    def get_id_token_credentials(self, target_audience: str = "") -> google.auth.credentials.Credentials:
+        connection_info = self.get_connection_info()
+        return get_id_token_credentials(
+            **connection_info,
+            scopes=self.scopes,
+            delegate_to=self.delegate_to,
+            target_audience=target_audience,
+        )
+
     def get_credentials(self) -> google.auth.credentials.Credentials:
         """Returns the Credentials object for Google API"""
         credentials, _ = self.get_credentials_and_project_id()

airflow/providers/google/provider.yaml

@@ -62,6 +62,7 @@ versions:
 dependencies:
   - apache-airflow>=2.3.0
   - apache-airflow-providers-common-sql>=1.3.1
+  - apache-airflow-providers-http>=4.0.0
   # Google has very clear rules on what dependencies should be used. All the limits below
   # follow strict guidelines of Google Libraries as quoted here:
   # While this issue is open, dependents of google-api-core, google-cloud-core. and google-auth
@@ -216,6 +217,11 @@ integrations:
       - /docs/apache-airflow-providers-google/operators/cloud/cloud_memorystore_memcached.rst
     logo: /integration-logos/gcp/Cloud-Memorystore.png
     tags: [gcp]
+  - integration-name: Google Cloud Run
+    external-doc-url: https://cloud.google.com/run/
+    how-to-guide:
+      - /docs/apache-airflow-providers-google/operators/cloud/cloud_run.rst
+    tags: [gcp]
   - integration-name: Google Cloud OS Login
     external-doc-url: https://cloud.google.com/compute/docs/oslogin/
     logo: /integration-logos/gcp/Google-Cloud-Generic.png
@@ -472,6 +478,9 @@ operators:
   - integration-name: Google Cloud Memorystore
     python-modules:
       - airflow.providers.google.cloud.operators.cloud_memorystore
+  - integration-name: Google Cloud Run
+    python-modules:
+      - airflow.providers.google.cloud.operators.cloud_run
   - integration-name: Google Cloud SQL
     python-modules:
       - airflow.providers.google.cloud.operators.cloud_sql
@@ -682,6 +691,9 @@ hooks:
   - integration-name: Google Cloud Memorystore
     python-modules:
       - airflow.providers.google.cloud.hooks.cloud_memorystore
+  - integration-name: Google Cloud Run
+    python-modules:
+      - airflow.providers.google.cloud.hooks.cloud_run
   - integration-name: Google Cloud SQL
     python-modules:
       - airflow.providers.google.cloud.hooks.cloud_sql