feat: add provider status info to landing page

Co-authored-by: Copilot <copilot@github.com>

Author: dalpasso

stac_fastapi/eodag/config.py

@@ -71,6 +71,13 @@ class Settings(ApiSettings):
         alias="validate",
     )
 
+    provider_online_status_threshold: float = Field(
+        default=0.8,
+        description="Threshold for considering a provider as online based"
+        "on the statuses of the all collections provided",
+        alias="provider_online_status_threshold",
+    )
+
 
 @lru_cache(maxsize=1)
 def get_settings() -> Settings:

stac_fastapi/eodag/dag.py

@@ -29,9 +29,10 @@
     TimeOutError,
 )
 from eodag.utils.requests import fetch_json
+from stac_fastapi.eodag.config import get_settings
 
 if TYPE_CHECKING:
-    from typing import Any
+    from typing import Any, Optional
 
     from fastapi import FastAPI
 
@@ -69,6 +70,73 @@ def fetch_external_stac_collections(
     return ext_stac_collections
 
 
+def get_providers_status(collections_list: CollectionsList) -> dict[str, dict[str, Any]]:
+    """Get the status of all providers based on their collections' status.
+
+    :param collections_list: list of EODAG collections
+    :return: Dictionary of providers' status indexed by provider ID
+    """
+    settings = get_settings()
+    online_threshold = settings.provider_online_status_threshold
+    # the key of providers_status is the provider ID
+    providers_status: dict[str, dict[str, Any]] = {}
+
+    def _get_last_check(status_1: dict[str, Any], status_2: dict[str, Any], key: str) -> Optional[str]:
+        """Get the most recent check date between two status dictionaries
+
+        The comparision handle possible `None` values in the dates.
+
+        :param status_1: First status dictionary
+        :param status_2: Second status dictionary
+        :param key: Key to check in the status dictionaries (e.g. "last_status_check" or "last_successful_check")
+        :return: Most recent check date as string, or None if not available in both dictionaries
+        """
+        check_1 = status_1.get(key)
+        check_2 = status_2.get(key)
+        if check_1 and check_2:
+            return max(check_1, check_2)
+        elif check_1:
+            return check_1
+        elif check_2:
+            return check_2
+        else:
+            return None
+
+    # count how many online/offline collections for each provider and keep the last check dates
+    for collection in collections_list:
+        if not getattr(collection, "federation", None):
+            continue
+        for provider_id, status in collection.federation.items():
+            d = {"online": 0, "offline": 0, "last_status_check": None, "last_successful_check": None}
+            ps = providers_status.setdefault(provider_id, d)
+            ps["last_status_check"] = _get_last_check(ps, status, "last_status_check")
+            ps["last_successful_check"] = _get_last_check(ps, status, "last_successful_check")
+            if status["status"] == "online":
+                ps["online"] += 1
+            else:
+                ps["offline"] += 1
+
+    # determine provider's status based on online/offline collections count and threshold
+    ret_providers_status = {}
+    for provider_id, status in providers_status.items():
+        online = status["online"]
+        offline = status["offline"]
+        total = online + offline
+        if total == 0:
+            provider_status = "offline"
+        elif online / total >= online_threshold:
+            provider_status = "online"
+        else:
+            provider_status = "offline"
+        ret_providers_status[provider_id] = {
+            "status": provider_status,
+            "last_status_check": status["last_status_check"],
+            "last_successful_check": status["last_successful_check"],
+        }
+
+    return ret_providers_status
+
+
 def init_dag(app: FastAPI) -> None:
     """Init EODataAccessGateway server instance, pre-running all time consuming tasks"""
     dag = EODataAccessGateway()
@@ -86,6 +154,9 @@ def init_dag(app: FastAPI) -> None:
 
     dag.db.upsert_collections(CollectionsDict.from_configs(collections))
 
+    # store providers status in app state
+    app.state.providers_status = get_providers_status(dag.list_collections())
+
     # pre-build search plugins
     for provider in dag.providers:
         next(dag._plugins_manager.get_search_plugins(provider=provider))

stac_fastapi/eodag/models/stac_metadata.py

@@ -211,10 +211,15 @@ def get_federation_backend_dict(request: Request, provider_name: str) -> dict[st
         for p in request.app.state.dag.providers.values()
         if provider_name in [p.name, p.metadata.get("group", None)]
     )
+    provider_status = next(
+        (status for p_id, status in request.app.state.providers_status.items() if provider.name == p_id),
+        {},
+    )
     return {
         "title": provider.metadata.get("group", None) or provider.name,
         "description": provider.metadata.get("description", None),
         "url": provider.metadata.get("url", None),
+        **provider_status,
     }