S3ToGCSOperator: Remove dest_gcs_conn_id (#23348)

Author: eladkal

airflow/providers/google/CHANGELOG.rst

@@ -132,6 +132,8 @@ Breaking changes
   ``GCSToLocalFilesystemOperator``
   ``LocalFilesystemToGCSOperator``
 
+* ``S3ToGCSOperator``: Remove ``dest_gcs_conn_id``. Please use ``gcp_conn_id``
+
 * ``BigQueryHook.create_empty_table`` Remove ``num_retries``. Please use ``retry``.
 
 * ``BigQueryHook.run_grant_dataset_view_access`` Remove ``source_project``. Please use ``project_id``.

airflow/providers/google/cloud/transfers/s3_to_gcs.py

@@ -15,7 +15,6 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-import warnings
 from tempfile import NamedTemporaryFile
 from typing import TYPE_CHECKING, Optional, Sequence, Union
 
@@ -57,8 +56,6 @@ class S3ToGCSOperator(S3ListOperator):
                  You can specify this argument if you want to use a different
                  CA cert bundle than the one used by botocore.
     :param gcp_conn_id: (Optional) The connection ID used to connect to Google Cloud.
-    :param dest_gcs_conn_id: (Deprecated) The connection ID used to connect to Google Cloud.
-        This parameter has been deprecated. You should pass the gcp_conn_id parameter instead.
     :param dest_gcs: The destination Google Cloud Storage bucket and prefix
         where you want to store the files. (templated)
     :param delegate_to: Google account to impersonate using domain-wide delegation of authority,
@@ -85,7 +82,7 @@ class S3ToGCSOperator(S3ListOperator):
            task_id="s3_to_gcs_example",
            bucket="my-s3-bucket",
            prefix="data/customers-201804",
-           dest_gcs_conn_id="google_cloud_default",
+           gcp_conn_id="google_cloud_default",
            dest_gcs="gs://my.gcs.bucket/some/customers/",
            replace=False,
            gzip=True,
@@ -114,7 +111,6 @@ def __init__(
         aws_conn_id='aws_default',
         verify=None,
         gcp_conn_id='google_cloud_default',
-        dest_gcs_conn_id=None,
         dest_gcs=None,
         delegate_to=None,
         replace=False,
@@ -124,16 +120,6 @@ def __init__(
     ):
 
         super().__init__(bucket=bucket, prefix=prefix, delimiter=delimiter, aws_conn_id=aws_conn_id, **kwargs)
-
-        if dest_gcs_conn_id:
-            warnings.warn(
-                "The dest_gcs_conn_id parameter has been deprecated. You should pass "
-                "the gcp_conn_id parameter.",
-                DeprecationWarning,
-                stacklevel=3,
-            )
-            gcp_conn_id = dest_gcs_conn_id
-
         self.gcp_conn_id = gcp_conn_id
         self.dest_gcs = dest_gcs
         self.delegate_to = delegate_to

tests/providers/google/cloud/transfers/test_s3_to_gcs.py

@@ -65,7 +65,7 @@ def test_execute(self, gcs_mock_hook, s3_one_mock_hook, s3_two_mock_hook):
             bucket=S3_BUCKET,
             prefix=S3_PREFIX,
             delimiter=S3_DELIMITER,
-            dest_gcs_conn_id=GCS_CONN_ID,
+            gcp_conn_id=GCS_CONN_ID,
             dest_gcs=GCS_PATH_PREFIX,
             google_impersonation_chain=IMPERSONATION_CHAIN,
         )
@@ -105,7 +105,7 @@ def test_execute_with_gzip(self, gcs_mock_hook, s3_one_mock_hook, s3_two_mock_ho
             bucket=S3_BUCKET,
             prefix=S3_PREFIX,
             delimiter=S3_DELIMITER,
-            dest_gcs_conn_id=GCS_CONN_ID,
+            gcp_conn_id=GCS_CONN_ID,
             dest_gcs=GCS_PATH_PREFIX,
             gzip=True,
         )