add form and constraints properties to Collection

cads_api_client/catalogue.py

@@ -39,7 +39,7 @@ def collection_ids(self) -> list[str]:
         ------
         list[str]
         """
-        return [collection["id"] for collection in self.json["collections"]]
+        return [collection["id"] for collection in self._json_dict["collections"]]
 
 
 @attrs.define
@@ -54,7 +54,7 @@ def begin_datetime(self) -> datetime.datetime | None:
         -------
         datetime.datetime or None
         """
-        if (value := self.json["extent"]["temporal"]["interval"][0][0]) is None:
+        if (value := self._json_dict["extent"]["temporal"]["interval"][0][0]) is None:
             return value
         return datetime.datetime.fromisoformat(value.replace("Z", "+00:00"))
 
@@ -66,7 +66,7 @@ def end_datetime(self) -> datetime.datetime | None:
         -------
         datetime.datetime or None
         """
-        if (value := self.json["extent"]["temporal"]["interval"][0][1]) is None:
+        if (value := self._json_dict["extent"]["temporal"]["interval"][0][1]) is None:
             return value
         return datetime.datetime.fromisoformat(value.replace("Z", "+00:00"))
 
@@ -78,7 +78,7 @@ def bbox(self) -> tuple[float, float, float, float]:
         -------
         tuple[float,float,float,float]
         """
-        return tuple(self.json["extent"]["spatial"]["bbox"][0])
+        return tuple(self._json_dict["extent"]["spatial"]["bbox"][0])
 
     @property
     def id(self) -> str:
@@ -88,7 +88,7 @@ def id(self) -> str:
         -------
         str
         """
-        return str(self.json["id"])
+        return str(self._json_dict["id"])
 
     @property
     def process(self) -> cads_api_client.Process:
@@ -102,6 +102,34 @@ def process(self) -> cads_api_client.Process:
         url = self._get_link_href(rel="retrieve")
         return cads_api_client.Process.from_request("get", url, **self._request_kwargs)
 
+    @property
+    def form(self) -> list[dict[str, Any]]:
+        """
+        Form JSON.
+
+        Returns
+        -------
+        list[dict[str,Any]]
+        """
+        url = f"{self.url}/form.json"
+        return ApiResponse.from_request(
+            "get", url, log_messages=False, **self._request_kwargs
+        )._json_list
+
+    @property
+    def constraints(self) -> list[dict[str, Any]]:
+        """
+        Constraints JSON.
+
+        Returns
+        -------
+        list[dict[str,Any]]
+        """
+        url = f"{self.url}/constraints.json"
+        return ApiResponse.from_request(
+            "get", url, log_messages=False, **self._request_kwargs
+        )._json_list
+
     def submit(self, **request: Any) -> cads_api_client.Remote:
         warnings.warn(
             "`.submit` has been deprecated, and in the future will raise an error."
@@ -156,7 +184,7 @@ def get_licenses(self, **params: Any) -> dict[str, Any]:
         response = ApiResponse.from_request(
             "get", url, params=params, **self._request_kwargs
         )
-        return response.json
+        return response._json_dict
 
     @property
     def messages(self) -> ApiResponse:

cads_api_client/legacy_api_client.py

@@ -215,7 +215,8 @@ def workflow(self, code, *args, **kwargs):
 
     def status(self, context: Any = None) -> dict[str, list[str]]:
         status = collections.defaultdict(list)
-        for message in self.client._catalogue_api.messages.json.get("messages", []):
+        messages = self.client._catalogue_api.messages._json_dict.get("messages", [])
+        for message in messages:
             status[message["severity"]].append(message["content"])
         return dict(status)
 

cads_api_client/processing.py

@@ -202,23 +202,35 @@ def url(self) -> str:
         return str(self.response.request.url)
 
     @property
-    def json(self) -> dict[str, Any]:
+    def json(self) -> Any:
         """Content of the response.
 
         Returns
         -------
-        dict[str,Any]
+        Any
         """
-        return dict(self.response.json())
+        return self.response.json()
+
+    @property
+    def _json_dict(self) -> dict[str, Any]:
+        assert isinstance(content := self.json, dict)
+        return content
+
+    @property
+    def _json_list(self) -> list[dict[str, Any]]:
+        assert isinstance(content := self.json, list)
+        return content
 
     def log_messages(self) -> None:
-        if message_str := self.json.get("message"):
+        if message_str := self._json_dict.get("message"):
             level, message_str = get_level_and_message(message_str)
             self.log(level, message_str)
 
-        messages = self.json.get("messages", [])
+        messages = self._json_dict.get("messages", [])
         dataset_messages = (
-            self.json.get("metadata", {}).get("datasetMetadata", {}).get("messages", [])
+            self._json_dict.get("metadata", {})
+            .get("datasetMetadata", {})
+            .get("messages", [])
         )
         for message_dict in messages + dataset_messages:
             if not (content := message_dict.get("content")):
@@ -231,7 +243,7 @@ def log_messages(self) -> None:
 
     def _get_links(self, rel: str | None = None) -> list[dict[str, str]]:
         links = []
-        for link in self.json.get("links", []):
+        for link in self._json_dict.get("links", []):
             if rel is not None and link.get("rel") == rel:
                 links.append(link)
         return links
@@ -304,7 +316,7 @@ def process_ids(self) -> list[str]:
         -------
         list[str]
         """
-        return [proc["id"] for proc in self.json["processes"]]
+        return [proc["id"] for proc in self._json_dict["processes"]]
 
 
 @attrs.define
@@ -319,7 +331,7 @@ def id(self) -> str:
         -------
         str
         """
-        process_id: str = self.json["id"]
+        process_id: str = self._json_dict["id"]
         return process_id
 
     def submit(self, **request: Any) -> cads_api_client.Remote:
@@ -361,7 +373,7 @@ def apply_constraints(self, **request: Any) -> dict[str, Any]:
             json={"inputs": request},
             **self._request_kwargs,
         )
-        return response.json
+        return response._json_dict
 
     def estimate_costs(self, **request: Any) -> dict[str, Any]:
         """Estimate costs of the parameters in a request.
@@ -382,7 +394,7 @@ def estimate_costs(self, **request: Any) -> dict[str, Any]:
             json={"inputs": request},
             **self._request_kwargs,
         )
-        return response.json
+        return response._json_dict
 
 
 @attrs.define(slots=False)
@@ -449,7 +461,7 @@ def json(self) -> dict[str, Any]:
         params = {"log": True, "request": True}
         if self.log_start_time:
             params["logStartTime"] = self.log_start_time
-        return self._get_api_response("get", params=params).json
+        return self._get_api_response("get", params=params)._json_dict
 
     @property
     def collection_id(self) -> str:
@@ -542,7 +554,7 @@ def results_ready(self) -> bool:
             return False
         if status == "failed":
             results = self.make_results(wait=False)
-            raise ProcessingFailedError(error_json_to_message(results.json))
+            raise ProcessingFailedError(error_json_to_message(results._json_dict))
         if status in ("dismissed", "deleted"):
             raise ProcessingFailedError(f"API state {status!r}")
         raise ProcessingFailedError(f"Unknown API state {status!r}")
@@ -584,7 +596,7 @@ def delete(self) -> dict[str, Any]:
         """
         response = self._get_api_response("delete")
         self.cleanup = False
-        return response.json
+        return response._json_dict
 
     def _warn(self) -> None:
         message = (
@@ -669,7 +681,7 @@ def request_uids(self) -> list[str]:
         -------
         list[str]
         """
-        return [job["jobID"] for job in self.json["jobs"]]
+        return [job["jobID"] for job in self._json_dict["jobs"]]
 
     @property
     def job_ids(self) -> list[str]:
@@ -699,7 +711,7 @@ def asset(self) -> dict[str, Any]:
         -------
         dict[str,Any]
         """
-        return dict(self.json["asset"]["value"])
+        return dict(self._json_dict["asset"]["value"])
 
     def _download(self, url: str, target: str) -> requests.Response:
         download_options = {"stream": True, "resume_transfers": True}

cads_api_client/profile.py

@@ -64,12 +64,14 @@ def _get_api_response(
 
     def accept_licence(self, licence_id: str, revision: int) -> dict[str, Any]:
         url = f"{self.url}/account/licences/{licence_id}"
-        return self._get_api_response("put", url, json={"revision": revision}).json
+        return self._get_api_response(
+            "put", url, json={"revision": revision}
+        )._json_dict
 
     def accepted_licences(self, **params: Any) -> dict[str, Any]:
         url = f"{self.url}/account/licences"
-        return self._get_api_response("get", url, params=params).json
+        return self._get_api_response("get", url, params=params)._json_dict
 
     def check_authentication(self) -> dict[str, Any]:
         url = f"{self.url}/account/verification/pat"
-        return self._get_api_response("post", url).json
+        return self._get_api_response("post", url)._json_dict

tests/integration_test_10_catalogue.py

@@ -81,3 +81,28 @@ def test_catalogue_get_licences(api_anon_client: ApiClient) -> None:
     assert portal_licences
     for portal_licence in portal_licences:
         assert portal_licence in all_licences
+
+
+def test_catalogue_form(collection: Collection) -> None:
+    assert set(collection.form[0]) == {
+        "css",
+        "default",
+        "details",
+        "help",
+        "id",
+        "label",
+        "name",
+        "required",
+        "type",
+    }
+
+
+def test_catalogue_constraints(collection: Collection) -> None:
+    assert set(collection.constraints[0]) == {
+        "day",
+        "month",
+        "product_type",
+        "time",
+        "variable",
+        "year",
+    }