feat(download): first class support for streaming download in eodag #2013

Author: pdavid-cssopra

docs/api_reference/eoproduct.rst

@@ -29,6 +29,7 @@ Download
 --------
 
 .. automethod:: EOProduct.download
+.. automethod:: EOProduct.stream_download
 .. automethod:: EOProduct.get_quicklook
 
 Conversion

docs/notebooks/tutos/tuto_stream_download_to_s3.ipynb

@@ -39,6 +39,7 @@ or run locally after being downloaded (see how to :ref:`install_notebooks`).
    notebooks/tutos/tuto_burnt_areas_snappy.ipynb
    notebooks/tutos/tuto_dedt_lumi_roi.ipynb
    notebooks/tutos/tuto_fedeo_ceda.ipynb
+   notebooks/tutos/tuto_stream_download_to_s3.ipynb
 
 .. grid:: 1 2 2 3
    :gutter: 4
@@ -124,3 +125,11 @@ or run locally after being downloaded (see how to :ref:`install_notebooks`).
       :shadow: md
 
       Access Fedeo data through the CEDA API using the dedicated EODAG plugin.
+
+   .. grid-item-card:: Download product as stream to S3
+      :link: notebooks/tutos/tuto_stream_download_to_s3
+      :link-type: doc
+      :text-align: center
+      :shadow: md
+
+      Download a product data as a stream and upload it on a S3 bucket without storing it locally.

docs/tutos.rst

@@ -190,6 +190,14 @@ def download(self, **kwargs: Unpack[DownloadConf]) -> str:
         """
         return self.product.download(asset=self.key, **kwargs)
 
+    def stream_download(self, **kwargs: Unpack[DownloadConf]) -> str:
+        """Downloads a single asset
+
+        :param kwargs: (optional) Additional named-arguments passed to `plugin.download()`
+        :returns: The absolute path to the downloaded product on the local filesystem
+        """
+        return self.product.stream_download(asset=self.key, **kwargs)
+
     def _repr_html_(self):
         thead = f"""<thead><tr><td style='text-align: left; color: grey;'>
             {type(self).__name__}&ensp;-&ensp;{self.key}

eodag/api/product/_assets.py

@@ -23,7 +23,7 @@
 import re
 import tempfile
 from datetime import datetime
-from typing import TYPE_CHECKING, Any, Iterable, Optional, Union, cast
+from typing import TYPE_CHECKING, Any, Iterable, Literal, Optional, Union, cast
 
 import orjson
 import requests
@@ -59,6 +59,7 @@
     STAC_VERSION,
     USER_AGENT,
     ProgressCallback,
+    StreamResponse,
     format_string,
     get_geometry_from_various,
 )
@@ -77,7 +78,6 @@
     from eodag.types.download_args import DownloadConf
     from eodag.utils import Unpack
 
-
 logger = logging.getLogger("eodag.product")
 
 
@@ -481,6 +481,50 @@ def download(
 
         return fs_path
 
+    def stream_download(
+        self,
+        byte_range: tuple[Optional[int], Optional[int]] = (None, None),
+        compress: Literal["zip", "raw", "auto"] = "auto",
+        wait: float = DEFAULT_DOWNLOAD_WAIT,
+        timeout: float = DEFAULT_DOWNLOAD_TIMEOUT,
+        **kwargs: Unpack[DownloadConf],
+    ) -> StreamResponse:
+        """Download as StreamResponse the EO product using the provided download plugin and the
+        authenticator if necessary.
+
+        :param byte_range: (optional) Tuple of first index / last index byte to read
+        :param compress: (optional) "zip", "raw", "auto"
+        :param wait: (optional) If download fails, wait time in minutes between
+                     two download tries
+        :param timeout: (optional) If download fails, maximum time in minutes
+                        before stop retrying to download
+        :param kwargs: additional kwargs like `dl_url_params` (dict) can be provided
+                        and will override any other values defined in a configuration
+                        file or with environment variables.
+        :returns: StreamResponse Stream representation of a file
+        :raises: :class:`~eodag.utils.exceptions.PluginImplementationError`
+        :raises: :class:`RuntimeError`
+        """
+        if self.downloader is None:
+            raise RuntimeError(
+                "EO product is unable to stream_download itself due to lacking of a "
+                "download plugin"
+            )
+        auth = (
+            self.downloader_auth.authenticate()
+            if self.downloader_auth is not None
+            else self.downloader_auth
+        )
+        return self.downloader.stream_download(
+            self,
+            auth,
+            byte_range,
+            compress,
+            wait=wait,
+            timeout=timeout,
+            **kwargs,
+        )
+
     def _init_progress_bar(
         self,
         progress_callback: Optional[ProgressCallback],

eodag/api/product/_product.py

@@ -16,3 +16,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """EODAG download package"""
+from .aws import AwsDownload
+from .base import Download
+from .http import HTTPDownload
+
+__all__ = ["Download", "AwsDownload", "HTTPDownload"]

eodag/plugins/download/__init__.py

@@ -37,7 +37,6 @@
     properties_from_xml,
 )
 from eodag.plugins.authentication.aws_auth import AwsAuth, raise_if_auth_error
-from eodag.plugins.download.base import Download
 from eodag.utils import (
     DEFAULT_DOWNLOAD_TIMEOUT,
     DEFAULT_DOWNLOAD_WAIT,
@@ -62,6 +61,8 @@
 )
 from eodag.utils.s3 import S3FileInfo, open_s3_zipped_object, stream_download_from_s3
 
+from .base import Download
+
 if TYPE_CHECKING:
     from mypy_boto3_s3 import S3ServiceResource
     from mypy_boto3_s3.client import S3Client
@@ -680,7 +681,7 @@ def _get_unique_products(
 
         return unique_product_chunks
 
-    def _stream_download_dict(
+    def stream_download(
         self,
         product: EOProduct,
         auth: Optional[Union[AuthBase, S3ServiceResource]] = None,
@@ -1176,3 +1177,6 @@ def get_chunk_dest_path(
 
         logger.debug(f"Downloading {chunk.key} to {product_path}")
         return product_path
+
+
+__all__ = ["AwsDownload"]

eodag/plugins/download/aws.py

@@ -24,6 +24,7 @@
 import tarfile
 import tempfile
 import zipfile
+from abc import abstractmethod
 from datetime import datetime, timedelta
 from pathlib import Path
 from time import sleep
@@ -102,6 +103,7 @@ def __init__(self, provider: str, config: PluginConfig) -> None:
         super(Download, self).__init__(provider, config)
         self._authenticate = bool(getattr(self.config, "authenticate", False))
 
+    @abstractmethod
     def download(
         self,
         product: EOProduct,
@@ -134,7 +136,8 @@ def download(
             "A Download plugin must implement a method named download"
         )
 
-    def _stream_download_dict(
+    @abstractmethod
+    def stream_download(
         self,
         product: EOProduct,
         auth: Optional[Union[AuthBase, S3ServiceResource]] = None,
@@ -145,7 +148,7 @@ def _stream_download_dict(
         **kwargs: Unpack[DownloadConf],
     ) -> StreamResponse:
         r"""
-        Base _stream_download_dict method. Not available, it must be defined for each plugin.
+        Base stream_download method. Not available, it must be defined for each plugin.
 
         :param product: The EO product to download
         :param auth: (optional) authenticated object
@@ -160,7 +163,7 @@ def _stream_download_dict(
         :returns: Dictionary of :class:`~fastapi.responses.StreamingResponse` keyword-arguments
         """
         raise NotImplementedError(
-            "Download streaming must be implemented using a method named _stream_download_dict"
+            "Download streaming must be implemented using a method named stream_download"
         )
 
     def _prepare_download(
@@ -766,3 +769,6 @@ def _config_executor(
 
         if thread_name_prefix:
             executor._thread_name_prefix = "eodag-download-all"
+
+
+__all__ = ["Download"]

eodag/plugins/download/base.py

@@ -662,7 +662,7 @@ def download_request(
             **kwargs: Unpack[DownloadConf],
         ) -> os.PathLike:
             is_empty = True
-            chunk_iterator = self._stream_download(
+            chunk_iterator = self._raw_stream_download(
                 product, auth, progress_callback, **kwargs
             )
             if fs_path is not None:
@@ -757,7 +757,7 @@ def _check_product_filename(self, product: EOProduct) -> str:
                         filename += ext
         return filename
 
-    def _stream_download_dict(
+    def stream_download(
         self,
         product: EOProduct,
         auth: Optional[Union[AuthBase, S3ServiceResource]] = None,
@@ -797,7 +797,7 @@ def _stream_download_dict(
             try:
                 assets_values = product.assets.get_values(kwargs.get("asset"))
                 with executor:
-                    assets_stream_list = self._stream_download_assets(
+                    assets_stream_list = self._raw_stream_download_assets(
                         product,
                         executor,
                         auth,
@@ -848,7 +848,7 @@ def _stream_download_dict(
                 else:
                     pass
 
-        chunk_iterator = self._stream_download(product, auth, None, **kwargs)
+        chunk_iterator = self._raw_stream_download(product, auth, None, **kwargs)
 
         # start reading chunks to set product.headers
         try:
@@ -957,7 +957,7 @@ def order(
             product, auth
         )
 
-    def _stream_download(
+    def _raw_stream_download(
         self,
         product: EOProduct,
         auth: Optional[AuthBase] = None,
@@ -1070,7 +1070,7 @@ def _stream_download(
             product.filename = filename
             return product._stream.iter_content(chunk_size=64 * 1024)
 
-    def _stream_download_assets(
+    def _raw_stream_download_assets(
         self,
         product: EOProduct,
         executor: ThreadPoolExecutor,
@@ -1247,7 +1247,7 @@ def _download_assets(
 
         assets_values = product.assets.get_values(kwargs.get("asset"))
 
-        assets_stream_list = self._stream_download_assets(
+        assets_stream_list = self._raw_stream_download_assets(
             product, executor, auth, progress_callback, assets_values, **kwargs
         )
 
@@ -1460,3 +1460,6 @@ def fetch_asset_size(asset: Asset) -> None:
             [f.result() for f in as_completed(futures)]
 
         return total_size
+
+
+__all__ = ["HTTPDownload"]