[ARC-3099] Propagate error details for invalid catalog search requests (#766)

* Propagate error details for invalid catalog search requests

* Clean up and finalize

Author: javidq

CHANGELOG.md

@@ -30,6 +30,10 @@ You can check your current version with the following command:
 
 For more information, see [UP42 Python package description](https://pypi.org/project/up42-py/).
 
+### 2.5.0a6
+**August 18, 2025**
+- Propagate request error message with `InvalidSearchRequest` when `Producer::search` fails.
+
 ### 2.5.0a5
 **August 15, 2025**
 - Fix how collection items are retrieved in `BulkDeletion`.

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "up42-py"
-version = "2.5.0a5"
+version = "2.5.0a6"
 description = "Python SDK for UP42, the geospatial marketplace and developer platform."
 authors = ["UP42 GmbH <[email protected]>"]
 license = "https://github.com/up42/up42-py/blob/master/LICENSE"

tests/test_glossary.py

@@ -3,6 +3,7 @@
 
 import geojson  # type: ignore
 import pytest
+import requests
 import requests_mock as req_mock
 
 from tests import constants, helpers
@@ -121,6 +122,7 @@ class TestProvider:
         description="description",
         roles=["PRODUCER", "HOST"],
     )
+    search_url = f"{constants.API_HOST}/catalog/hosts/{HOST_NAME}/stac/search"
 
     @pytest.mark.parametrize(
         "provider, is_host",
@@ -137,6 +139,28 @@ def test_fails_to_search_if_provider_is_not_host(self):
         with pytest.raises(glossary.InvalidHost):
             next(dataclasses.replace(self.provider, roles=["PRODUCER"]).search())
 
+    def test_fails_to_search_if_search_request_is_invalid(self, requests_mock: req_mock.Mocker):
+        error_message = "invalid request"
+        requests_mock.post(
+            url=self.search_url,
+            status_code=422,
+            json={
+                "data": {},
+                "error": {"code": 422, "message": error_message, "details": "ignored"},
+            },
+        )
+        with pytest.raises(glossary.InvalidSearchRequest, match=error_message):
+            next(self.provider.search())
+
+    @pytest.mark.parametrize("error_code", [400, 401, 403, 500])
+    def test_should_propagate_search_failures(self, error_code: int, requests_mock: req_mock.Mocker):
+        requests_mock.post(
+            url=self.search_url,
+            status_code=error_code,
+        )
+        with pytest.raises(requests.HTTPError):
+            next(self.provider.search())
+
     @pytest.mark.parametrize("bbox", [None, BBOX])
     @pytest.mark.parametrize("intersects", [None, POLYGON])
     @pytest.mark.parametrize(
@@ -179,10 +203,9 @@ def test_should_search(
         if collections:
             search_params["collections"] = collections
 
-        search_url = f"{constants.API_HOST}/catalog/hosts/{self.provider.name}/stac/search"
-        next_page_url = f"{search_url}/next"
+        next_page_url = f"{self.search_url}/next"
         requests_mock.post(
-            url=search_url,
+            url=self.search_url,
             json={
                 "type": "FeatureCollection",
                 "features": [SCENE_FEATURE] * 3,

up42/glossary.py

@@ -3,6 +3,7 @@
 from typing import Any, Iterator, Literal, Optional, Union
 
 import geojson  # type: ignore
+import requests
 
 from up42 import base, host, utils
 
@@ -65,6 +66,10 @@ class InvalidHost(ValueError):
     pass
 
 
+class InvalidSearchRequest(ValueError):
+    pass
+
+
 @dataclasses.dataclass
 class Provider:
     session = base.Session()
@@ -112,12 +117,18 @@ def search(
         def get_pages():
             url = host.endpoint(f"/catalog/hosts/{self.name}/stac/search")
             while url:
-                page: dict = self.session.post(url, json=payload).json()
-                yield page["features"]
-                url = next(
-                    (link["href"] for link in page["links"] if link["rel"] == "next"),
-                    None,
-                )
+                try:
+                    page: dict = self.session.post(url, json=payload).json()
+                    yield page["features"]
+                    url = next(
+                        (link["href"] for link in page["links"] if link["rel"] == "next"),
+                        None,
+                    )
+                except requests.HTTPError as http_error:
+                    if http_error.response.status_code == 422:
+                        error = http_error.response.json()["error"]
+                        raise InvalidSearchRequest(error["message"]) from http_error
+                    raise http_error
 
         for page in get_pages():
             for feature in page: