feat(asset): move download_link property to an asset #2029

eodag/api/product/_assets.py

@@ -17,10 +17,15 @@
 # limitations under the License.
 from __future__ import annotations
 
+import logging
 import re
 from collections import UserDict
-from typing import TYPE_CHECKING, Any, Optional
+from typing import TYPE_CHECKING, Any, Optional, Union
 
+from typing_extensions import override
+
+from eodag.api.product.metadata_mapping import NOT_AVAILABLE, OFFLINE_STATUS
+from eodag.utils import guess_file_type
 from eodag.utils.exceptions import NotAvailableError
 from eodag.utils.repr import dict_to_html_table
 
@@ -29,6 +34,8 @@
     from eodag.types.download_args import DownloadConf
     from eodag.utils import Unpack
 
+logger = logging.getLogger("eodag.api.product.assets")
+
 
 class AssetsDict(UserDict):
     """A UserDict object which values are :class:`~eodag.api.product._assets.Asset`
@@ -56,12 +63,84 @@ class AssetsDict(UserDict):
 
     product: EOProduct
 
+    TECHNICAL_ASSETS = ["download_link", "quicklook", "thumbnail"]
+
     def __init__(self, product: EOProduct, *args: Any, **kwargs: Any) -> None:
         self.product = product
         super(AssetsDict, self).__init__(*args, **kwargs)
 
     def __setitem__(self, key: str, value: dict[str, Any]) -> None:
+        if not self._check(key, value):
+            return
         super().__setitem__(key, Asset(self.product, key, value))
+        self.sort()
+
+    @override
+    def update(self, value: Union[dict[str, Any], AssetsDict]) -> None:  # type: ignore
+        """Used to self update with exernal value"""
+        buffer: dict = {}
+        for key in value:
+            if self._check(key, value[key]):
+                buffer[key] = value[key]
+        super().update(buffer)
+        self.sort()
+
+    def _check(self, asset_key: str, asset_values: dict[str, Any]) -> bool:
+
+        # Asset must have href or order_link
+        href = asset_values.pop("href", None)
+        if href not in [None, "", NOT_AVAILABLE]:
+            asset_values["href"] = href
+        order_link = asset_values.pop("order_link", None)
+        if order_link not in [None, "", NOT_AVAILABLE]:
+            asset_values["order_link"] = order_link
+
+        if "href" not in asset_values and "order_link" not in asset_values:
+            logger.warning(
+                "asset '{}' skipped ignored because neither href nor order_link is available".format(
+                    asset_key
+                ),
+            )
+            return False
+
+        def target_url(asset: dict) -> Optional[str]:
+            target_url = None
+            if "href" in asset:
+                target_url = asset["href"]
+            elif "order_link" in asset:
+                target_url = asset["order_link"]
+            return target_url
+
+        assets = self.as_dict()
+        used_urls = []
+        for key in assets:
+            if key == asset_key:
+                # Duplicated key
+                return False
+            used_urls.append(target_url(assets[key]))
+
+        # Prevent asset key / asset target url replication (out from technical ones)
+        # thumbnail and quicklook can share same url
+        url = target_url(asset_values)
+        if asset_key not in AssetsDict.TECHNICAL_ASSETS and (url in used_urls):
+            # Duplicated url
+            return False
+
+        return True
+
+    def sort(self):
+        """Used to self sort"""
+        sorted_assets = {}
+        data = self.as_dict()
+        # Keep technical assets first
+        for key in AssetsDict.TECHNICAL_ASSETS:
+            if key in data:
+                sorted_assets[key] = data.pop(key)
+        # Sort and add others
+        data = dict(sorted(data.items()))
+        for key in data:
+            sorted_assets[key] = data[key]
+        super().update(sorted_assets)
 
     def as_dict(self) -> dict[str, Any]:
         """Builds a representation of AssetsDict to enable its serialization
@@ -165,14 +244,55 @@ class Asset(UserDict):
     """
 
     product: EOProduct
+
+    # Location
+    location: Optional[str]
+    remote_location: Optional[str]
+
+    # File
     size: int
     filename: Optional[str]
     rel_path: str
 
     def __init__(self, product: EOProduct, key: str, *args: Any, **kwargs: Any) -> None:
         self.product = product
         self.key = key
+        self.location = None
+        self.remote_location = None
         super(Asset, self).__init__(*args, **kwargs)
+        self._update()
+
+    def __setitem__(self, key, item):
+        super().__setitem__(key, item)
+        self._update()
+
+    def _update(self):
+
+        title = self.get("title", None)
+        if title is None:
+            super().__setitem__("title", self.key)
+
+        # Order link behaviour require order:status state
+        orderlink = self.get("order_link", None)
+        orderstatus = self.get("order:status", None)
+        if orderlink is not None and orderstatus is None:
+            super().__setitem__("order:status", OFFLINE_STATUS)
+
+        href = self.get("href", None)
+        if href is None:
+            super().__setitem__("href", "")
+            href = ""
+
+        if href != "":
+            # Provide location and remote_location when undefined and href updated
+            if self.location is None:
+                self.location = href
+            if self.remote_location is None:
+                self.remote_location = href
+            # With order behaviour, href can be fill later
+            content_type = self.get("type", None)
+            if content_type is None:
+                super().__setitem__("type", guess_file_type(href))
 
     def as_dict(self) -> dict[str, Any]:
         """Builds a representation of Asset to enable its serialization

eodag/api/product/metadata_mapping.py

@@ -24,6 +24,7 @@
 from datetime import datetime, timedelta
 from string import Formatter
 from typing import TYPE_CHECKING, Any, AnyStr, Callable, Iterator, Optional, Union, cast
+from urllib.parse import unquote
 
 import geojson
 import orjson
@@ -189,6 +190,8 @@ def format_metadata(search_param: str, *args: Any, **kwargs: Any) -> str:
         - ``to_rounded_wkt``: simplify the WKT of a geometry
         - ``to_title``: Convert a string to title case
         - ``to_upper``: Convert a string to uppercase
+        - ``url_decode``: Convert a string url_encoded to decoded ones
+        - ``round``: Convert a string number to another one without decimal part
 
     :param search_param: The string to be formatted
     :param args: (optional) Additional arguments to use in the formatting process
@@ -782,6 +785,28 @@ def convert_to_upper(string: str) -> str:
             """Convert a string to uppercase."""
             return string.upper()
 
+        @staticmethod
+        def convert_url_decode(string: str):
+            return unquote(string)
+
+        @staticmethod
+        def convert_round(string: Any) -> str:
+            """Convert a number string to integer string"""
+            if isinstance(string, float):
+                return str(int(string))
+            elif isinstance(string, int):
+                return str(string)
+            elif isinstance(string, str):
+                formatted = ""
+                for i in range(0, len(string)):
+                    char = string[i]
+                    if char == ".":
+                        break
+                    if "0123456789".find(char) >= 0:
+                        formatted += char
+                return formatted
+            return string
+
         @staticmethod
         def convert_to_title(string: str) -> str:
             """Convert a string to title case."""
@@ -1518,10 +1543,17 @@ def format_query_params(
 
         if COMPLEX_QS_REGEX.match(provider_search_param):
             parts = provider_search_param.split("=")
+
             if len(parts) == 1:
-                formatted_query_param = format_metadata(
-                    provider_search_param, collection, **query_dict
-                )
+
+                # If part contains something to interprete
+                if parts[0].strip("{}").find("{") >= 0:
+                    formatted_query_param = format_metadata(
+                        provider_search_param, collection, **query_dict
+                    )
+                else:
+                    formatted_query_param = "{" + parts[0].strip("{}") + "}"
+
                 formatted_query_param = formatted_query_param.replace("'", '"')
                 if "{{" in provider_search_param:
                     # retrieve values from hashes where keys are given in the param

eodag/config.py

@@ -580,7 +580,7 @@ class MetadataPreMapping(TypedDict, total=False):
 
     def __or__(self, other: Union[Self, dict[str, Any]]) -> Self:
         """Return a new PluginConfig with merged values."""
-        new_config = self.__class__.from_mapping(self.__dict__)
+        new_config: Self = self.__class__.from_mapping(self.__dict__)
         new_config.update(other)
         return new_config
 

eodag/plugins/apis/usgs.py

@@ -134,6 +134,8 @@ def authenticate(self) -> None:
                     os.remove(api.TMPFILE)
                     continue
                 raise AuthenticationError("Please check your USGS credentials.") from e
+            except Exception as e:
+                raise AuthenticationError("Authentication failure") from e
 
     def query(
         self,