Google Cloud Providers - Fix _MethodDefault deepcopy failure

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

@@ -18,6 +18,8 @@
 """This module contains a Google API base operator."""
 from __future__ import annotations
 
+from google.api_core.gapic_v1.method import DEFAULT
+
 from airflow.models import BaseOperator
 
 
@@ -27,4 +29,11 @@ class GoogleCloudBaseOperator(BaseOperator):
     on top of Google API client libraries.
     """
 
-    pass
+    def __deepcopy__(self, memo):
+        """
+        Updating the memo to fix the non-copyable global constant.
+        This constant can be specified in operator parameters as a retry configuration to indicate a default.
+        See https://github.com/apache/airflow/issues/28751 for details.
+        """
+        memo[id(DEFAULT)] = DEFAULT
+        return super().__deepcopy__(memo)

tests/providers/google/cloud/operators/test_cloud_base.py

@@ -0,0 +1,56 @@
+#
+# 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.
+from __future__ import annotations
+
+import copy
+import unittest
+
+from google.api_core.gapic_v1.method import DEFAULT, _MethodDefault
+from google.api_core.retry import Retry
+
+from airflow.providers.google.cloud.operators.cloud_base import GoogleCloudBaseOperator
+
+TASK_ID = "task-id"
+
+
+class GoogleSampleOperator(GoogleCloudBaseOperator):
+    def __init__(
+        self,
+        retry: Retry | _MethodDefault = DEFAULT,
+        config: dict | None = None,
+        **kwargs,
+    ) -> None:
+        super().__init__(**kwargs)
+        self.retry = retry
+        self.config = config
+
+
+class TestGoogleCloudBaseOperator(unittest.TestCase):
+    def test_handles_deepcopy_with_method_default(self):
+        op = GoogleSampleOperator(task_id=TASK_ID)
+        copied_op = copy.deepcopy(op)
+
+        self.assertEqual(copied_op.retry, DEFAULT)
+        self.assertEqual(copied_op.config, None)
+
+    def test_handles_deepcopy_with_non_default_retry(self):
+        op = GoogleSampleOperator(task_id=TASK_ID, retry=Retry(deadline=30), config={"config": "value"})
+        copied_op = copy.deepcopy(op)
+
+        self.assertEqual(copied_op.retry.deadline, 30)
+        self.assertEqual(copied_op.config, {"config": "value"})