feat(core): move download_link property to an asset (#2091)

Co-authored-by: Sylvain Brunato <sylvain.brunato@c-s.fr>

Author: pdavid-cssopra

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_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,10 @@
     from eodag.types.download_args import DownloadConf
     from eodag.utils import StreamResponse, Unpack
 
+logger = logging.getLogger("eodag.api.product.assets")
+
+TECHNICAL_ASSET_KEYS: tuple[str, ...] = ("download_link", "quicklook", "thumbnail")
+
 
 class AssetsDict(UserDict):
     """A UserDict object which values are :class:`~eodag.api.product._assets.Asset`
@@ -47,11 +56,12 @@ class AssetsDict(UserDict):
     ...     provider="foo",
     ...     properties={"id": "bar", "geometry": "POINT (0 0)"}
     ... )
-    >>> type(product.assets)
-    <class 'eodag.api.product._assets.AssetsDict'>
+    >>> from eodag.api.product._assets import AssetsDict
+    >>> isinstance(product.assets, AssetsDict)
+    True
     >>> product.assets.update({"foo": {"href": "http://somewhere/something"}})
     >>> product.assets
-    {'foo': {'href': 'http://somewhere/something'}}
+    {'foo': {'href': 'http://somewhere/something', 'title': 'foo', 'type': 'application/octet-stream'}}
     """
 
     product: EOProduct
@@ -60,12 +70,97 @@ def __init__(self, product: EOProduct, *args: Any, **kwargs: Any) -> None:
         self.product = product
         super(AssetsDict, self).__init__(*args, **kwargs)
 
-    def update(self, data: dict[str, Any]) -> None:  # type: ignore
-        """Update assets"""
-        super().update(data)
-
     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()
+        self._update_product_location()
+
+    @override
+    def update(self, *args: Any, **kwargs: Any) -> None:
+        """Used to self update with external value"""
+        incoming = {k: v for k, v in dict(*args, **kwargs).items() if self._check(k, v)}
+
+        if not incoming:
+            return
+
+        super().update(incoming)
+        self.sort()
+        self._update_product_location()
+
+    def _update_product_location(self) -> None:
+        """Backward-compat: propagate ``download_link`` asset to product location.
+
+        ``EOProduct.location`` / ``EOProduct.remote_location`` are still expected to
+        carry the download URL by downstream code (download plugins, server-mode, ...).
+        Technical assets are NOT written back to ``EOProduct.properties``.
+        """
+        dl = self.data.get("download_link")
+        if dl is None:
+            return
+        href = dl.get("href") or ""
+        if href:
+            if not self.product.location:
+                self.product.location = href
+            if not self.product.remote_location:
+                self.product.remote_location = href
+
+    def _check(self, asset_key: str, asset_value: dict[str, Any]) -> bool:
+        """Validate an asset before insertion.
+
+        Checks that the asset has a valid ``href`` or ``order_link`` and that its
+        URL is not already used by another (non-technical) asset.  Mutates
+        *asset_value* in place by stripping empty/unavailable href/order_link entries.
+
+        :param asset_key: Key under which the asset will be stored
+        :param asset_value: Mutable asset dictionary (modified in place)
+        :returns: ``True`` if the asset is valid and can be inserted, ``False`` otherwise
+        """
+        # Asset must have href or order_link
+        href = asset_value.pop("href", None)
+        if href not in [None, "", NOT_AVAILABLE]:
+            asset_value["href"] = href
+        order_link = asset_value.pop("order_link", None)
+        if order_link not in [None, "", NOT_AVAILABLE]:
+            asset_value["order_link"] = order_link
+
+        if "href" not in asset_value and "order_link" not in asset_value:
+            logger.warning(
+                "asset '{}' skipped ignored because neither href nor order_link is available".format(
+                    asset_key
+                ),
+            )
+            return False
+
+        def target_url(asset: dict[str, Any]) -> Optional[str]:
+            return asset.get("href") or asset.get("order_link")
+
+        assets = self.as_dict()
+        used_urls = [
+            target_url(asset) for key, asset in assets.items() if key != asset_key
+        ]
+
+        # Prevent asset key / asset target url replication (out from technical ones)
+        # thumbnail and quicklook can share same url
+        url = target_url(asset_value)
+        if asset_key not in TECHNICAL_ASSET_KEYS and (url in used_urls):
+            # Duplicated url
+            return False
+
+        return True
+
+    def sort(self):
+        """Used to self sort"""
+        sorted_assets = {}
+        # Keep technical assets first
+        for key in TECHNICAL_ASSET_KEYS:
+            if key in self.data:
+                sorted_assets[key] = self.data.pop(key)
+        # Sort and add others
+        for key in sorted(self.data):
+            sorted_assets[key] = self.data[key]
+        self.data = sorted_assets
 
     def as_dict(self) -> dict[str, Any]:
         """Builds a representation of AssetsDict to enable its serialization
@@ -83,29 +178,31 @@ def get_values(self, asset_filter: str = "", regex=True) -> list[Asset]:
         :param regex: Uses regex to match the asset key or simply compare strings
         :return: list of assets
         """
-        if asset_filter:
-            if regex:
-                filter_regex = re.compile(asset_filter)
-                assets_keys = list(self.keys())
-                assets_keys = list(filter(filter_regex.fullmatch, assets_keys))
-            else:
-                assets_keys = [a for a in self.keys() if a == asset_filter]
-            filtered_assets = {}
-            if len(assets_keys) > 0:
-                filtered_assets = {a_key: self.get(a_key) for a_key in assets_keys}
-            assets_values = [a for a in filtered_assets.values() if a and "href" in a]
-            if not assets_values and regex:
-                # retry without regex
-                return self.get_values(asset_filter, regex=False)
-            elif not assets_values:
-                raise NotAvailableError(
-                    rf"No asset key matching re.fullmatch(r'{asset_filter}') was found in {self.product}"
-                )
-            else:
-                return assets_values
-        else:
+        if not asset_filter:
             return [a for a in self.values() if "href" in a]
 
+        if regex:
+            filter_regex = re.compile(asset_filter)
+            assets_keys = [k for k in self.keys() if filter_regex.fullmatch(k)]
+        else:
+            assets_keys = [k for k in self.keys() if k == asset_filter]
+
+        assets_values = [
+            a
+            for k in assets_keys
+            for a in [self.get(k)]
+            if isinstance(a, Asset) and "href" in a
+        ]
+
+        if assets_values:
+            return assets_values
+        if regex:
+            # retry without regex
+            return self.get_values(asset_filter, regex=False)
+        raise NotAvailableError(
+            rf"No asset key matching re.fullmatch(r'{asset_filter}') was found in {self.product}"
+        )
+
     def _repr_html_(self, embeded=False):
         thead = (
             f"""<thead><tr><td style='text-align: left; color: grey;'>
@@ -162,21 +259,69 @@ class Asset(UserDict):
     ...     properties={"id": "bar", "geometry": "POINT (0 0)"}
     ... )
     >>> product.assets.update({"foo": {"href": "http://somewhere/something"}})
-    >>> type(product.assets["foo"])
-    <class 'eodag.api.product._assets.Asset'>
+    >>> from eodag.api.product._assets import Asset
+    >>> isinstance(product.assets["foo"], Asset)
+    True
     >>> product.assets["foo"]
-    {'href': 'http://somewhere/something'}
+    {'href': 'http://somewhere/something', 'title': 'foo', 'type': 'application/octet-stream'}
     """
 
     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):
+        """Normalize asset fields and sync ``location``/``remote_location`` from ``href``.
+
+        Fills in default ``title``, ``href``, ``order:status`` and ``type`` entries
+        when missing, and mirrors a non-empty ``href`` into the asset's
+        ``location`` / ``remote_location`` attributes.
+        """
+
+        title = self.get("title")
+        if title is None:
+            super().__setitem__("title", self.key)
+
+        # Order link behaviour require order:status state
+        orderlink = self.get("order_link")
+        orderstatus = self.get("order:status")
+        if orderlink is not None and orderstatus is None:
+            super().__setitem__("order:status", OFFLINE_STATUS)
+
+        href = self.get("href")
+        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")
+            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 @@
 import re
 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
@@ -39,7 +40,6 @@
 from shapely.geometry import LineString, MultiPolygon, Point, Polygon
 from shapely.ops import transform
 
-from eodag.api.product._assets import Asset
 from eodag.types.queryables import Queryables
 from eodag.utils import (
     DEFAULT_PROJ,
@@ -63,6 +63,7 @@
 
     from shapely.geometry.base import BaseGeometry
 
+    from eodag.api.product._assets import Asset
     from eodag.config import PluginConfig
 
 logger = logging.getLogger("eodag.product.metadata_mapping")
@@ -211,6 +212,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
@@ -832,6 +835,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."""
@@ -1607,10 +1632,18 @@ 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 (nested braces or a converter)
+                inner = parts[0].strip("{}")
+                if inner.find("{") >= 0 or "#" in inner:
+                    formatted_query_param = format_metadata(
+                        provider_search_param, collection, **query_dict
+                    )
+                else:
+                    formatted_query_param = "{" + inner + "}"
+
                 formatted_query_param = formatted_query_param.replace("'", '"')
                 if "{{" in provider_search_param:
                     # retrieve values from hashes where keys are given in the param
@@ -2049,6 +2082,7 @@ def normalize_bands(data: Union[dict, Asset]) -> Union[dict, Asset]:
         "statistics",
     ]
     EXCLUDE_MOVE_TO_PARENT_BAND_FIELDNAME = ["name", "eo:common_name"]
+    from eodag.api.product._assets import Asset
 
     # https://github.com/radiantearth/stac-spec/blob/v1.1.0/best-practices.md#bands
     # Migrate band STAC 1.0 to 1.1

eodag/config.py

@@ -586,7 +586,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/ecmwf.py

@@ -242,7 +242,7 @@ def download(
             raise DownloadError(e)
 
         with open(record_filename, "w") as fh:
-            fh.write(product.properties["eodag:download_link"])
+            fh.write(product.remote_location)
         logger.debug("Download recorded in %s", record_filename)
 
         # do not try to extract a directory