Repository: apache/airflow
Pull request: Add new operator gdrive to local
I have added new operator, but I noticed that there are no return value in execute function. I understand that this is not finished version, but I need help and review and what can be next step.

closes: #10126 


<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.

Feel free to ping committers for the review!

In case of existing issue, reference it using one of the following:

closes: #ISSUE
related: #ISSUE

How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->

---
**^ Add meaningful description above**

Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/master/UPDATING.md).


Problem statement:
Add download function at gdrive hook
**Description**

Right now the GoogleDriveHook class from: airflow/airflow/providers/google/suite/hooks/drive.py has only a upload function, it would be nice if a download function was also provided.

**Use case / motivation**

This is useful for example if a lot of data is keep in drive in some xlsx file and you want to obtain that data, automatically as a source for a ETL process.

Hints from issue discussion:
Thanks for opening your first issue here! Be sure to follow the issue template!

@SaladBreaker would you like to implement it? I'm happy to help with it, let me know if you need any guidance 🚀 
Thanks, I'll try it.
any update on this?
Hello, can I implement this feature?
@Scuall1992 assigned to you.
Just mentioning that we have also https://github.com/apache/airflow/issues/13966 but download isn't part of it.
Hello, can you help me? As I understood I need to develop download function in hooks/drive.py, with these signature

def download_file(self, remote_location: str, local_location: str) -> str:

I'm receiving file id and download it in local_location.

If file isn't exists then what type of exception I should raise?
Also I found the difference between downloading Google Doc files and others https://developers.google.com/drive/api/v3/manage-downloads

What we need in this issue?

Note that the `GoogleDriveHook` inherits from the [GoogleBaseHook](https://airflow.apache.org/docs/apache-airflow-providers-google/stable/_modules/airflow/providers/google/common/hooks/base_google.html#GoogleBaseHook), which has the method `download_content_from_request`. You can use this method to write data to a file handle, e.g. a `BytesIO` object. You can then write the contents of that object to disk.

Instead of adding functionality to the hook, you might want to consider creating a transfer operator, e.g. `gdrive_to_local.py` and leverage the existing functionality of `GoogleDriveHook`. You can use the `_upload_data` method in the [GoogleDriveToGCSOperator](https://github.com/RNHTTR/airflow/blob/f9d442f9a85b4b705bbe574f718034cf6bab5486/airflow/providers/google/cloud/transfers/gdrive_to_gcs.py) as an example, but instead of passing the value of the file handle to GCS, you can [write it to a file](https://stackoverflow.com/a/60450462/7004653).

Commit message: Add gdrive to local test

Patch to review:
diff --git a/airflow/providers/google/cloud/example_dags/example_gdrive_to_local.py b/airflow/providers/google/cloud/example_dags/example_gdrive_to_local.py
new file mode 100644
index 0000000000000..94fca91105231
--- /dev/null
+++ b/airflow/providers/google/cloud/example_dags/example_gdrive_to_local.py
@@ -0,0 +1,49 @@
+#
+# 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.
+
+import os
+
+from airflow import models
+from airflow.providers.google.cloud.transfers.gdrive_to_local import GoogleDriveToLocalOperator
+from airflow.providers.google.suite.sensors.drive import GoogleDriveFileExistenceSensor
+from airflow.utils.dates import days_ago
+
+FOLDER_ID = os.environ.get("FILE_ID", "1234567890qwerty")
+FILE_NAME = os.environ.get("FILE_NAME", "file.pdf")
+OUTPUT_FILE = os.environ.get("OUTPUT_FILE", "out_file.pdf")
+
+with models.DAG(
+    "example_gdrive_to_local_with_gdrive_sensor",
+    start_date=days_ago(1),
+    schedule_interval=None,  # Override to match your needs
+    tags=["example"],
+) as dag:
+    # [START detect_file]
+    detect_file = GoogleDriveFileExistenceSensor(
+        task_id="detect_file", folder_id=FOLDER_ID, file_name=FILE_NAME
+    )
+    # [END detect_file]
+    # [START download_from_gdrive_to_local]
+    download_from_gdrive_to_local = GoogleDriveToLocalOperator(
+        task_id="download_from_gdrive_to_local",
+        folder_id=FOLDER_ID,
+        file_name=FILE_NAME,
+        output_file=OUTPUT_FILE,
+    )
+    # [END download_from_gdrive_to_local]
+    detect_file >> download_from_gdrive_to_local
diff --git a/airflow/providers/google/cloud/transfers/gdrive_to_local.py b/airflow/providers/google/cloud/transfers/gdrive_to_local.py
new file mode 100644
index 0000000000000..b03c8dc5ac754
--- /dev/null
+++ b/airflow/providers/google/cloud/transfers/gdrive_to_local.py
@@ -0,0 +1,95 @@
+# 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 typing import Optional, Sequence, Union
+
+from airflow.models import BaseOperator
+from airflow.providers.google.suite.hooks.drive import GoogleDriveHook
+from airflow.utils.decorators import apply_defaults
+
+
+class GoogleDriveToLocalOperator(BaseOperator):
+    """
+    Writes a Google Drive file into local Storage.
+
+    .. seealso::
+        For more information on how to use this operator, take a look at the guide:
+        :ref:`howto/operator:GoogleDriveToLocalOperator`
+
+    :param output_file: Path to downloaded file
+    :type output_file: str
+    :param folder_id: The folder id of the folder in which the Google Drive file resides
+    :type folder_id: str
+    :param file_name: The name of the file residing in Google Drive
+    :type file_name: str
+    :param drive_id: Optional. The id of the shared Google Drive in which the file resides.
+    :type drive_id: str
+    :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
+        domain-wide delegation enabled.
+    :type delegate_to: str
+    :param impersonation_chain: Optional service account to impersonate using short-term
+        credentials, or chained list of accounts required to get the access_token
+        of the last account in the list, which will be impersonated in the request.
+        If set as a string, the account must grant the originating account
+        the Service Account Token Creator IAM role.
+        If set as a sequence, the identities from the list must grant
+        Service Account Token Creator IAM role to the directly preceding identity, with first
+        account from the list granting this role to the originating account (templated).
+    :type impersonation_chain: Union[str, Sequence[str]]
+    """
+
+    template_fields = [
+        "output_file",
+        "folder_id",
+        "file_name",
+        "drive_id",
+        "impersonation_chain",
+    ]
+
+    @apply_defaults
+    def __init__(
+        self,
+        *,
+        output_file: str,
+        file_name: str,
+        folder_id: str,
+        drive_id: Optional[str] = None,
+        delegate_to: Optional[str] = None,
+        impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
+        **kwargs,
+    ) -> None:
+        super().__init__(**kwargs)
+        self.output_file = output_file
+        self.folder_id = folder_id
+        self.drive_id = drive_id
+        self.file_name = file_name
+        self.delegate_to = delegate_to
+        self.impersonation_chain = impersonation_chain
+
+    def execute(self, context):
+        self.log.info('Executing download: %s into %s', self.file_name, self.output_file)
+        gdrive_hook = GoogleDriveHook(
+            delegate_to=self.delegate_to,
+            impersonation_chain=self.impersonation_chain,
+        )
+        file_metadata = gdrive_hook.get_file_id(
+            folder_id=self.folder_id, file_name=self.file_name, drive_id=self.drive_id
+        )
+
+        with open(self.output_file, "wb") as file:
+            gdrive_hook.download_file(file_id=file_metadata["id"], file_handle=file)
diff --git a/airflow/providers/google/provider.yaml b/airflow/providers/google/provider.yaml
index 5222acc14f2b1..95eb58cc78948 100644
--- a/airflow/providers/google/provider.yaml
+++ b/airflow/providers/google/provider.yaml
@@ -699,6 +699,10 @@ transfers:
     target-integration-name: Local
     how-to-guide: /docs/apache-airflow-providers-google/operators/transfer/gcs_to_local.rst
     python-module: airflow.providers.google.cloud.transfers.gcs_to_local
+  - source-integration-name: Google Drive
+    target-integration-name: Local
+    how-to-guide: /docs/apache-airflow-providers-google/operators/transfer/gdrive_to_local.rst
+    python-module: airflow.providers.google.cloud.transfers.gdrive_to_local
   - source-integration-name: Salesforce
     target-integration-name: Google Cloud Storage (GCS)
     how-to-guide: /docs/apache-airflow-providers-google/operators/transfer/salesforce_to_gcs.rst
diff --git a/airflow/providers/google/suite/hooks/drive.py b/airflow/providers/google/suite/hooks/drive.py
index 6e27c71f98202..ce713484611ba 100644
--- a/airflow/providers/google/suite/hooks/drive.py
+++ b/airflow/providers/google/suite/hooks/drive.py
@@ -22,6 +22,7 @@
 from googleapiclient.http import HttpRequest, MediaFileUpload
 
 from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
+from io import TextIOWrapper
 
 
 class GoogleDriveHook(GoogleBaseHook):
@@ -203,13 +204,13 @@ def upload_file(self, local_location: str, remote_location: str) -> str:
         :rtype: str
         """
         service = self.get_conn()
-        directory_path, _, filename = remote_location.rpartition("/")
+        directory_path, _, file_name = remote_location.rpartition("/")
         if directory_path:
             parent = self._ensure_folders_exists(directory_path)
         else:
             parent = "root"
 
-        file_metadata = {"name": filename, "parents": [parent]}
+        file_metadata = {"name": file_name, "parents": [parent]}
         media = MediaFileUpload(local_location)
         file = (
             service.files()  # pylint: disable=no-member
@@ -218,3 +219,15 @@ def upload_file(self, local_location: str, remote_location: str) -> str:
         )
         self.log.info("File %s uploaded to gdrive://%s.", local_location, remote_location)
         return file.get("id")
+
+    def download_file(self, file_id: str, file_handle: TextIOWrapper, chunk_size: int = 104857600):
+        """
+        Download a file from Google Drive.
+
+        :param file_id: the id of the file
+        :type file_id: str
+        :param file_handle: file handle used to write the content to
+        :type file_handle: io.TextIOWrapper
+        """
+        request = self.get_media_request(file_id=file_id)
+        self.download_content_from_request(file_handle=file_handle, request=request, chunk_size=chunk_size)
diff --git a/docs/apache-airflow-providers-google/operators/transfer/gdrive_to_local.rst b/docs/apache-airflow-providers-google/operators/transfer/gdrive_to_local.rst
new file mode 100644
index 0000000000000..6a725710e8d98
--- /dev/null
+++ b/docs/apache-airflow-providers-google/operators/transfer/gdrive_to_local.rst
@@ -0,0 +1,58 @@
+ .. 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.
+
+
+Downloads data from Google Drive Storage to Local Filesystem
+============================================================
+The `Google Drive <https://www.google.com/drive/>`__ is
+used to store daily use data, including documents and photos. Google Drive has built-in mechanisms to facilitate group work e.g.
+document editor, file sharing mechanisms.
+
+.. contents::
+  :depth: 1
+  :local:
+
+
+Prerequisite Tasks
+^^^^^^^^^^^^^^^^^^
+
+.. include::/operators/_partials/prerequisite_tasks.rst
+
+.. _howto/operator:GoogleDriveToLocalOperator:
+
+GCSToLocalFilesystemOperator
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+:class:`~airflow.providers.google.cloud.transfers.gdrive_to_local.GoogleDriveToLocalOperator` allows you to download
+data from Google Drive to local filesystem.
+
+
+Below is an example of using this operator to download file from Google Drive to Local Filesystem.
+
+.. exampleinclude:: /../../airflow/providers/google/cloud/example_dags/example_gdrive_to_local.py
+    :language: python
+    :dedent: 4
+    :start-after: [START download_from_gdrive_to_local]
+    :end-before: [END download_from_gdrive_to_local]
+
+
+Reference
+---------
+
+For further information, look at:
+
+* `Google Drive API Documentation <https://developers.google.com/drive/api/v3/about-sdk>`__
diff --git a/tests/providers/google/cloud/transfers/test_gdrive_to_local.py b/tests/providers/google/cloud/transfers/test_gdrive_to_local.py
new file mode 100644
index 0000000000000..65635bb518c3e
--- /dev/null
+++ b/tests/providers/google/cloud/transfers/test_gdrive_to_local.py
@@ -0,0 +1,38 @@
+#
+# 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 unittest import TestCase, mock
+
+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"
+OUTPUT_FILE = "out_file.pdf"
+
+
+class TestGoogleDriveToLocalOperator(TestCase):
+    @mock.patch("airflow.providers.google.cloud.transfers.gdrive_to_local.GoogleDriveHook")
+    def test_execute(self, hook_mock):
+        op = GoogleDriveToLocalOperator(
+            task_id=TASK_ID,
+            folder_id=FOLDER_ID,
+            file_name=FILE_NAME,
+            output_file=OUTPUT_FILE,
+        )
+        op.execute(context=None)
+        hook_mock.assert_called_once()


Code review comments:
diff_hunk: @@ -0,0 +1,38 @@
+#
+# 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 unittest import TestCase, mock
+
+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"
+OUTPUT_FILE = "out_file.pdf"
+
+
+class TestGoogleDriveToLocalOperator(TestCase):
+    @mock.patch("airflow.providers.google.cloud.transfers.gdrive_to_local.GoogleDriveHook")
+    def test_execute(self, hook_mock):
+        op = GoogleDriveToLocalOperator(
+            task_id=TASK_ID,
+            folder_id=FOLDER_ID,
+            file_name=FILE_NAME,
+            output_file=OUTPUT_FILE,
+        )
+        op.execute(context=None)
+        hook_mock.assert_called_once()
path: tests/providers/google/cloud/transfers/test_gdrive_to_local.py
text: @user1:
```suggestion
        hook_mock.assert_called_once_with(delegate_to=None, impersonation_chain=None)
```
This only verify instantiating the hook. We also need to check if right method was called so we need:
```py
hook_mock.return_value.get_file_id.assert_called_once_with(...)
hook_mock.return_value.download_file.assert_called_once_with(...)
```
where the `...` needs to be replaced with proper values. Also consider using
```py
with TemporaryNamedFile("wb") as temp_file:
    op = GoogleDriveToLocalOperator(..., output_file=temp_file.name)
```
to avoid a side effect in form of a created file.