Support YAML input for CloudBuildCreateOperator

airflow/providers/google/cloud/example_dags/example_cloud_build.py

@@ -99,6 +99,12 @@
         task_id="create_build_from_repo_result",
     )
 
+    create_build_from_file = CloudBuildCreateOperator(
+        task_id="create_build_from_file", project_id=GCP_PROJECT_ID, body='example_cloud_build.yaml'
+    )
+
     create_build_from_storage >> create_build_from_storage_result  # pylint: disable=pointless-statement
 
     create_build_from_repo >> create_build_from_repo_result  # pylint: disable=pointless-statement
+
+    create_build_from_file  # pylint: disable=pointless-statement

airflow/providers/google/cloud/example_dags/example_cloud_build.yaml

@@ -0,0 +1,20 @@
+# 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.
+
+steps:
+  - name: 'ubuntu'
+    args: ['echo', 'hello world']

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

@@ -75,7 +75,7 @@ def create_build(self, body: Dict, project_id: str) -> Dict:
         Starts a build with the specified configuration.
 
         :param body: The request body.
-            See: https://cloud.google.com/cloud-build/docs/api/reference/rest/Shared.Types/Build
+            See: https://cloud.google.com/cloud-build/docs/api/reference/rest/v1/projects.builds
         :type body: dict
         :param project_id: Optional, Google Cloud Project project_id where the function belongs.
             If set to None or missing, the default project_id from the GCP connection is used.

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

@@ -18,9 +18,11 @@
 """Operators that integrat with Google Cloud Build service."""
 import re
 from copy import deepcopy
-from typing import Any, Dict, Iterable, Optional
+from typing import Any, Dict, Iterable, Optional, Union
 from urllib.parse import unquote, urlparse
 
+import yaml
+
 from airflow.exceptions import AirflowException
 from airflow.models import BaseOperator
 from airflow.providers.google.cloud.hooks.cloud_build import CloudBuildHook
@@ -39,7 +41,7 @@ class BuildProcessor:
     * It is possible to provide the source as the URL address instead dict.
 
     :param body: The request body.
-        See: https://cloud.google.com/cloud-build/docs/api/reference/rest/Shared.Types/Build
+        See: https://cloud.google.com/cloud-build/docs/api/reference/rest/v1/projects.builds
     :type body: dict
     """
     def __init__(self, body: Dict) -> None:
@@ -83,13 +85,29 @@ def _reformat_storage_source(self):
 
         self.body["source"]["storageSource"] = self._convert_storage_url_to_dict(source)
 
+    def _load_body_to_dict(self):
+        """
+        :param file:
+            file path to YAML build config
+        :return: dict
+        """
+        try:
+            with open(self.body, 'r') as f:
+                self.body = yaml.safe_load(f)
+        except yaml.YAMLError as e:
+            raise AirflowException("Exception when loading resource definition: %s\n" % e)
+
     def process_body(self):
         """
         Processes the body passed in the constructor
 
         :return: the body.
         :type: dict
         """
+
+        if isinstance(self.body, str):
+            self._load_body_to_dict()
+            return self.body
         self._verify_source()
         self._reformat_source()
         return self.body
@@ -178,7 +196,7 @@ class CloudBuildCreateOperator(BaseOperator):
 
     @apply_defaults
     def __init__(self,
-                 body: dict,
+                 body: Union[dict, str],
                  project_id: Optional[str] = None,
                  gcp_conn_id: str = "google_cloud_default",
                  api_version: str = "v1",