Source code for

# -*- coding: utf-8 -*-
# Copyright 2018, CS GROUP - France,
# This file is part of EODAG project
# Licensed 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
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations

import hashlib
import logging
import os
import os.path
from typing import TYPE_CHECKING, Any, Dict, Optional, Union
from xml.dom import minidom
from xml.parsers.expat import ExpatError

import requests
from requests import RequestException

from eodag.api.product.metadata_mapping import OFFLINE_STATUS, ONLINE_STATUS
from import Download
from import HTTPDownload
from eodag.utils import (
from eodag.utils.exceptions import (

    from eodag.api.product import EOProduct
    from eodag.config import PluginConfig

logger = logging.getLogger("")

[docs]class S3RestDownload(Download): """Http download on S3-like object storage location for example using Mundi REST API (free account) Re-use AwsDownload bucket some handling methods :param provider: provider name :type provider: str :param config: Download plugin configuration: * ``config.base_uri`` (str) - default endpoint url * ``config.extract`` (bool) - (optional) extract downloaded archive or not * ``config.auth_error_code`` (int) - (optional) authentication error code * ``config.bucket_path_level`` (int) - (optional) bucket location index in path.split('/') * ``config.order_enabled`` (bool) - (optional) wether order is enabled or not if product is `OFFLINE` * ``config.order_method`` (str) - (optional) HTTP request method, GET (default) or POST * ``config.order_headers`` (dict) - (optional) order request headers * ``config.order_on_response`` (dict) - (optional) edit or add new product properties * ``config.order_status_method`` (str) - (optional) status HTTP request method, GET (default) or POST * ``config.order_status_percent`` (str) - (optional) progress percentage key in obtained status response * ``config.order_status_success`` (dict) - (optional) key/value identifying an error success * ``config.order_status_on_success`` (dict) - (optional) edit or add new product properties :type config: :class:`~eodag.config.PluginConfig` """
[docs] def __init__(self, provider: str, config: PluginConfig) -> None: super(S3RestDownload, self).__init__(provider, config) self.http_download_plugin = HTTPDownload(self.provider, self.config)
def download( self, product: EOProduct, auth: Optional[PluginConfig] = None, progress_callback: Optional[ProgressCallback] = None, wait: int = DEFAULT_DOWNLOAD_WAIT, timeout: int = DEFAULT_DOWNLOAD_TIMEOUT, **kwargs: Union[str, bool, Dict[str, Any]], ) -> Optional[str]: """Download method for S3 REST API. :param product: The EO product to download :type product: :class:`~eodag.api.product._product.EOProduct` :param auth: (optional) The configuration of a plugin of type Authentication :type auth: :class:`~eodag.config.PluginConfig` :param progress_callback: (optional) A method or a callable object which takes a current size and a maximum size as inputs and handle progress bar creation and update to give the user a feedback on the download progress :type progress_callback: :class:`~eodag.utils.ProgressCallback` or None :param kwargs: `outputs_prefix` (str), `extract` (bool), `delete_archive` (bool) and `dl_url_params` (dict) can be provided as additional kwargs and will override any other values defined in a configuration file or with environment variables. :type kwargs: Union[str, bool, dict] :returns: The absolute path to the downloaded product in the local filesystem :rtype: str """ if progress_callback is None: "Progress bar unavailable, please call instead of" ) progress_callback = ProgressCallback(disable=True) # order product if it is offline ordered_message = "" if ( "orderLink" in and "storageStatus" in and["storageStatus"] != ONLINE_STATUS ): self.http_download_plugin.orderDownload(product=product, auth=auth) @self._download_retry(product, wait, timeout) def download_request( product: EOProduct, auth: PluginConfig, progress_callback: ProgressCallback, ordered_message: str, **kwargs: Any, ): # check order status if"orderStatusLink", None): self.http_download_plugin.orderDownloadStatus( product=product, auth=auth ) # get bucket urls bucket_name, prefix = get_bucket_name_and_prefix( url=product.location, bucket_path_level=self.config.bucket_path_level ) if ( bucket_name is None and "storageStatus" in and["storageStatus"] == OFFLINE_STATUS ): raise NotAvailableError( "%s is not available for download on %s (status = %s)" % (["title"], self.provider,["storageStatus"], ) ) bucket_url = urljoin( product.downloader.config.base_uri.strip("/") + "/", bucket_name ) nodes_list_url = bucket_url + "?prefix=" + prefix.strip("/") # get nodes/files list contained in the bucket logger.debug("Retrieving product content from %s", nodes_list_url) bucket_contents = requests.get( nodes_list_url, auth=auth, headers=USER_AGENT, timeout=HTTP_REQ_TIMEOUT ) try: bucket_contents.raise_for_status() except requests.RequestException as err: # check if error is identified as auth_error in provider conf auth_errors = getattr(self.config, "auth_error_code", [None]) if not isinstance(auth_errors, list): auth_errors = [auth_errors] if err.response and err.response.status_code in auth_errors: raise AuthenticationError( "HTTP Error %s returned, %s\nPlease check your credentials for %s" % ( err.response.status_code, err.response.text.strip(), self.provider, ) ) # product not available elif ("storageStatus", ONLINE_STATUS) != ONLINE_STATUS ): msg = ( ordered_message if ordered_message and err.response and not err.response.text.strip() else err.response and err.response.text.strip() ) raise NotAvailableError( "%s(initially %s) requested, returned: %s" % (["title"],["storageStatus"], msg, ) ) # other error else: logger.exception( "Could not get content from %s (provider:%s, plugin:%s)\n%s", nodes_list_url, self.provider, self.__class__.__name__, bucket_contents.text, ) raise RequestError(str(err)) try: xmldoc = minidom.parseString(bucket_contents.text) except ExpatError as err: logger.exception("Could not parse xml data from %s", bucket_contents) raise DownloadError(str(err)) nodes_xml_list = xmldoc.getElementsByTagName("Contents") if len(nodes_xml_list) == 0: logger.warning("Could not load any content from %s", nodes_list_url) # destination product path outputs_prefix = ( kwargs.pop("outputs_prefix", None) or self.config.outputs_prefix ) abs_outputs_prefix = os.path.abspath(outputs_prefix) product_local_path = os.path.join(abs_outputs_prefix, prefix.split("/")[-1]) # .downloaded cache record directory download_records_dir = os.path.join(abs_outputs_prefix, ".downloaded") try: os.makedirs(download_records_dir) except OSError as exc: import errno if exc.errno != errno.EEXIST: # Skip error if dir exists import traceback as tb logger.warning( "Unable to create records directory. Got:\n%s", tb.format_exc() ) # check if product has already been downloaded url_hash = hashlib.md5(product.remote_location.encode("utf-8")).hexdigest() record_filename = os.path.join(download_records_dir, url_hash) if os.path.isfile(record_filename) and os.path.exists(product_local_path): product.location = path_to_uri(product_local_path) return product_local_path # Remove the record file if product_local_path is absent (e.g. it was deleted while record wasn't) elif os.path.isfile(record_filename): logger.debug( "Record file found (%s) but not the actual file", record_filename ) logger.debug("Removing record file : %s", record_filename) os.remove(record_filename) # total size for progress_callback total_size = sum( [ int(node.firstChild.nodeValue) for node in xmldoc.getElementsByTagName("Size") ] ) progress_callback.reset(total=total_size) # download each node key for node_xml in nodes_xml_list: node_key = unquote( node_xml.getElementsByTagName("Key")[0].firstChild.nodeValue ) # As "Key", "Size" and "ETag" (md5 hash) can also be retrieved from node_xml node_url = urljoin(bucket_url.strip("/") + "/", node_key.strip("/")) # output file location local_filename_suffix_list = node_key.split("/")[6:] if local_filename_suffix_list[0] == os.path.basename( product_local_path ): local_filename_suffix_list.pop(0) # single file: remove nested sub dirs if len(nodes_xml_list) == 1: local_filename_suffix_list = [local_filename_suffix_list[-1]] local_filename = os.path.join( product_local_path, *local_filename_suffix_list ) local_filename_dir = os.path.dirname(os.path.realpath(local_filename)) if not os.path.isdir(local_filename_dir): os.makedirs(local_filename_dir) with requests.get( node_url, stream=True, auth=auth, headers=USER_AGENT, timeout=DEFAULT_STREAM_REQUESTS_TIMEOUT, ) as stream: try: stream.raise_for_status() except RequestException: import traceback as tb logger.error( "Error while getting resource :\n%s", tb.format_exc() ) else: with open(local_filename, "wb") as fhandle: for chunk in stream.iter_content(chunk_size=64 * 1024): if chunk: fhandle.write(chunk) progress_callback(len(chunk)) with open(record_filename, "w") as fh: fh.write(product.remote_location) logger.debug("Download recorded in %s", record_filename) product.location = path_to_uri(product_local_path) return product_local_path return download_request( product, auth, progress_callback, ordered_message, **kwargs, )