Drop duplicate build_raw copy from CloudBuildCreateBuildOperator

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

@@ -190,8 +190,6 @@ def __init__(
     ) -> None:
         super().__init__(**kwargs)
         self.build = build
-        # Not template fields to keep original value
-        self.build_raw = build
         self.project_id = project_id
         self.wait = wait
         self.retry = retry
@@ -205,12 +203,14 @@ def __init__(
 
     def prepare_template(self) -> None:
         # if no file is specified, skip
-        if not isinstance(self.build_raw, str):
+        if not isinstance(self.build, str):
             return
-        with open(self.build_raw) as file:
-            if self.build_raw.endswith((".yaml", ".yml")):
+        # Keep the path in a local: the branches below replace self.build with a dict.
+        build_path = self.build
+        with open(build_path) as file:
+            if build_path.endswith((".yaml", ".yml")):
                 self.build = yaml.safe_load(file.read())
-            if self.build_raw.endswith(".json"):
+            if build_path.endswith(".json"):
                 self.build = json.loads(file.read())
 
     @property

providers/google/tests/unit/google/cloud/operators/test_cloud_build.py

@@ -165,6 +165,28 @@ def test_load_templated(self, file_type, file_content):
             expected_body = {"steps": [{"name": "ubuntu", "args": ["echo", "Hello {{ params.name }}!"]}]}
             assert expected_body == operator.build
 
+    def test_init_does_not_duplicate_build(self):
+        build_path = "path/to/build.json"
+        operator = CloudBuildCreateBuildOperator(build=build_path, task_id="task-id")
+        assert operator.build == build_path
+        # Any second attribute referencing the same object would be a copy prepare_template()
+        # could read instead of self.build, whatever it is named.
+        assert [name for name, value in vars(operator).items() if value is build_path] == ["build"]
+
+    def test_prepare_template_second_call_is_no_op(self, tmp_path):
+        expected_body = {"steps": [{"name": "ubuntu", "args": ["echo", "Hello {{ params.name }}!"]}]}
+        build_file = tmp_path / "build.json"
+        build_file.write_text(json.dumps(expected_body))
+
+        operator = CloudBuildCreateBuildOperator(build=str(build_file), task_id="task-id")
+        operator.prepare_template()
+        assert expected_body == operator.build
+
+        build_file.unlink()
+
+        operator.prepare_template()
+        assert expected_body == operator.build
+
     @mock.patch(CLOUD_BUILD_HOOK_PATH)
     def test_create_build_trigger(self, mock_hook):
         mock_hook.return_value.create_build_trigger.return_value = BuildTrigger()

scripts/ci/prek/validate_operators_init_exemptions.txt

@@ -7,5 +7,4 @@
 # Burn-down tracked at https://github.com/apache/airflow/issues/70296
 providers/amazon/src/airflow/providers/amazon/aws/operators/neptune.py::NeptuneStartDbClusterOperator
 providers/amazon/src/airflow/providers/amazon/aws/operators/neptune.py::NeptuneStopDbClusterOperator
-providers/google/src/airflow/providers/google/cloud/operators/cloud_build.py::CloudBuildCreateBuildOperator
 providers/google/src/airflow/providers/google/cloud/operators/dataproc.py::DataprocCreateClusterOperator