Add gcp_conn_id argument to GoogleDriveToLocalOperator

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

@@ -35,6 +35,7 @@ class GoogleDriveToLocalOperator(BaseOperator):
     :param output_file: Path to downloaded file
     :param folder_id: The folder id of the folder in which the Google Drive file resides
     :param file_name: The name of the file residing in Google Drive
+    :param gcp_conn_id: The GCP connection ID to use when fetching connection info.
     :param drive_id: Optional. The id of the shared Google Drive in which the file resides.
     :param delegate_to: The account to impersonate using domain-wide delegation of authority,
         if any. For this to work, the service account making the request must have
@@ -64,6 +65,7 @@ def __init__(
         file_name: str,
         folder_id: str,
         drive_id: Optional[str] = None,
+        gcp_conn_id: str = "google_cloud_default",
         delegate_to: Optional[str] = None,
         impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
         **kwargs,
@@ -73,12 +75,14 @@ def __init__(
         self.folder_id = folder_id
         self.drive_id = drive_id
         self.file_name = file_name
+        self.gcp_conn_id = gcp_conn_id
         self.delegate_to = delegate_to
         self.impersonation_chain = impersonation_chain
 
     def execute(self, context: 'Context'):
         self.log.info('Executing download: %s into %s', self.file_name, self.output_file)
         gdrive_hook = GoogleDriveHook(
+            gcp_conn_id=self.gcp_conn_id,
             delegate_to=self.delegate_to,
             impersonation_chain=self.impersonation_chain,
         )

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

@@ -18,28 +18,36 @@
 from tempfile import NamedTemporaryFile
 from unittest import TestCase, mock
 
-from airflow.providers.google.cloud.transfers.gdrive_to_local import GoogleDriveToLocalOperator
+from airflow.providers.google.cloud.transfers.gdrive_to_local import (
+    GoogleDriveToLocalOperator,
+)
 
 TASK_ID = "test-drive-to-local-operator"
 FOLDER_ID = "1234567890qwerty"
 FILE_NAME = "file.pdf"
+GCP_CONN_ID = "google_cloud_default"
 
 
 class TestGoogleDriveToLocalOperator(TestCase):
-    @mock.patch("airflow.providers.google.cloud.transfers.gdrive_to_local.GoogleDriveHook")
+    @mock.patch(
+        "airflow.providers.google.cloud.transfers.gdrive_to_local.GoogleDriveHook"
+    )
     def test_execute(self, hook_mock):
         with NamedTemporaryFile("wb") as temp_file:
             op = GoogleDriveToLocalOperator(
                 task_id=TASK_ID,
                 folder_id=FOLDER_ID,
                 file_name=FILE_NAME,
+                gcp_conn_id=GCP_CONN_ID,
                 output_file=temp_file.name,
             )
             meta = {"id": "123xyz"}
             hook_mock.return_value.get_file_id.return_value = meta
 
             op.execute(context=None)
-            hook_mock.assert_called_once_with(delegate_to=None, impersonation_chain=None)
+            hook_mock.assert_called_once_with(
+                delegate_to=None, gcp_conn_id=GCP_CONN_ID, impersonation_chain=None
+            )
 
             hook_mock.return_value.download_file.assert_called_once_with(
                 file_id=meta["id"], file_handle=mock.ANY