Update results href (#203)

* update results href

* update

* update protocol stripping

Author: mcucchi9

cads_processing_api_service/config.py

@@ -17,6 +17,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import os
+import random
+
 import pydantic_settings
 
 API_REQUEST_TEMPLATE = """import cdsapi
@@ -64,10 +67,20 @@ class Settings(pydantic_settings.BaseSettings):
     anonymous_licences_message: str = ANONYMOUS_LICENCES_MESSAGE
     deprecation_warning_message: str = DEPRECATION_WARNING_MESSAGE
 
+    download_nodes_config: str = "/etc/retrieve-api/download-nodes.config"
+
     @property
     def profiles_api_url(self) -> str:
         return f"http://{self.profiles_service}:{self.profiles_api_service_port}"
 
+    @property
+    def data_volume(self) -> str:
+        data_volumes_config_path = self.download_nodes_config
+        with open(data_volumes_config_path) as fp:
+            data_volumes = [os.path.expandvars(line.rstrip("\n")) for line in fp]
+        data_volume = random.choice(data_volumes)
+        return data_volume
+
 
 def ensure_settings(
     settings: Settings | None = None,

cads_processing_api_service/utils.py

@@ -18,6 +18,7 @@
 import datetime
 import enum
 import threading
+import urllib.parse
 from typing import Any, Callable, Mapping
 
 import cachetools
@@ -469,6 +470,14 @@ def get_job_from_broker_db(
     return job
 
 
+def update_results_href(local_path: str, data_volume: str | None = None) -> str:
+    if data_volume is None:
+        data_volume = config.ensure_settings().data_volume
+    file_path = local_path.split("://", 1)[-1]
+    results_href = urllib.parse.urljoin(data_volume, file_path)
+    return results_href
+
+
 def get_results_from_job(
     job: cads_broker.SystemRequest, session: sqlalchemy.orm.Session
 ) -> dict[str, Any]:
@@ -496,11 +505,12 @@ def get_results_from_job(
     if job_status == "successful":
         try:
             asset_value = job.cache_entry.result["args"][0]  # type: ignore
-            results = {"asset": {"value": asset_value}}
         except Exception:
             raise exceptions.JobResultsExpired(
                 detail=f"results of job {job_id} expired"
             )
+        asset_value["href"] = update_results_href(asset_value["file:local_path"])
+        results = {"asset": {"value": asset_value}}
     elif job_status == "failed":
         error_messages = get_job_events(
             job=job, session=session, event_type="user_visible_error"

tests/test_30_utils.py

@@ -243,19 +243,47 @@ def test_get_job_from_broker_db() -> None:
             job = utils.get_job_from_broker_db("1234", session=mock_session)
 
 
+def test_update_results_href() -> None:
+    data_volume = "http://data_volume/"
+
+    local_path = "protocol://results/1234"
+    updated_href = utils.update_results_href(local_path, data_volume)
+    exp_updated_href = "http://data_volume/results/1234"
+    assert updated_href == exp_updated_href
+
+    local_path = "results/1234"
+    updated_href = utils.update_results_href(local_path, data_volume)
+    exp_updated_href = "http://data_volume/results/1234"
+    assert updated_href == exp_updated_href
+
+
 def test_get_results_from_job() -> None:
     mock_session = unittest.mock.Mock(spec=sqlalchemy.orm.Session)
     job = cads_broker.SystemRequest(
         **{
             "status": "successful",
             "request_uid": "1234",
             "cache_entry": cacholote.database.CacheEntry(
-                result={"args": [{"key": "value"}]}
+                result={
+                    "args": [{"key": "value", "file:local_path": "test_local_path"}]
+                }
             ),
         }
     )
-    results = utils.get_results_from_job(job, session=mock_session)
-    exp_results = {"asset": {"value": {"key": "value"}}}
+    with unittest.mock.patch(
+        "cads_processing_api_service.utils.update_results_href"
+    ) as mock_update_results_href:
+        mock_update_results_href.return_value = "test_href"
+        results = utils.get_results_from_job(job, session=mock_session)
+    exp_results = {
+        "asset": {
+            "value": {
+                "key": "value",
+                "file:local_path": "test_local_path",
+                "href": "test_href",
+            }
+        }
+    }
     assert results == exp_results
 
     job = cads_broker.SystemRequest(