Validate JSON response shapes

Author: adamtheturtle

src/vws/_json_utils.py

@@ -0,0 +1,69 @@
+"""Validation helpers for JSON received from remote services."""
+
+import json
+from typing import TypeGuard
+
+
+def _is_json_object(value: object, /) -> TypeGuard[dict[str, object]]:
+    """Return whether a decoded JSON value is an object."""
+    return isinstance(value, dict)
+
+
+def _is_object_list(value: object, /) -> TypeGuard[list[object]]:
+    """Return whether a decoded JSON value is an array."""
+    return isinstance(value, list)
+
+
+def _validated_object(*, value: object) -> dict[str, object]:
+    """Return a decoded JSON object."""
+    if not _is_json_object(value):
+        msg = "Expected a JSON object."
+        raise TypeError(msg)
+    return value
+
+
+def json_object(*, value: str | bytes | bytearray) -> dict[str, object]:
+    """Decode and validate a JSON object."""
+    loaded: object = json.loads(s=value)
+    return _validated_object(value=loaded)
+
+
+def string_value(*, value: object, name: str) -> str:
+    """Return a JSON value after validating that it is a string."""
+    if not isinstance(value, str):
+        msg = f"{name} must be a string."
+        raise TypeError(msg)
+    return value
+
+
+def string_field(*, value: dict[str, object], name: str) -> str:
+    """Return a required string field from a JSON object."""
+    return string_value(value=value[name], name=name)
+
+
+def string_list_field(
+    *,
+    value: dict[str, object],
+    name: str,
+) -> list[str]:
+    """Return a required list of strings from a JSON object."""
+    items = value[name]
+    if not _is_object_list(items) or not all(
+        isinstance(item, str) for item in items
+    ):
+        msg = f"{name} must be a list of strings."
+        raise TypeError(msg)
+    return [item for item in items if isinstance(item, str)]
+
+
+def object_list_field(
+    *,
+    value: dict[str, object],
+    name: str,
+) -> list[dict[str, object]]:
+    """Return a required list of JSON objects."""
+    items = value[name]
+    if not _is_object_list(items):
+        msg = f"{name} must be a list of JSON objects."
+        raise TypeError(msg)
+    return [_validated_object(value=item) for item in items]

src/vws/async_query.py

@@ -12,6 +12,7 @@
 
 from vws._image_utils import ImageType as _ImageType
 from vws._image_utils import get_image_data as _get_image_data
+from vws._json_utils import json_object, object_list_field, string_field
 from vws.exceptions.base_exceptions import CloudRecoError
 from vws.exceptions.cloud_reco_exceptions import (
     AuthenticationFailureError,
@@ -201,13 +202,16 @@ async def query(
             raise CloudRecoError(response=response)
 
         try:
-            response_body = json.loads(s=response.text)
+            response_body = json_object(value=response.text)
         except json.JSONDecodeError as exc:
             if response.status_code >= HTTPStatus.BAD_REQUEST:
                 raise CloudRecoError(response=response) from exc
             raise
 
-        result_code = response_body["result_code"]  # pyrefly: ignore [unknown-variable-type]
+        result_code = string_field(
+            value=response_body,
+            name="result_code",
+        )
         if result_code != "Success":
             exception = {
                 "AuthenticationFailure": (AuthenticationFailureError),
@@ -217,8 +221,11 @@ async def query(
             }[result_code]
             raise exception(response=response)
 
-        result_list = list(response_body["results"])  # pyrefly: ignore [unknown-argument-type]
+        result_list = object_list_field(
+            value=response_body,
+            name="results",
+        )
         return [
-            QueryResult.from_response_dict(response_dict=item)  # pyrefly: ignore [unknown-argument-type]
+            QueryResult.from_response_dict(response_dict=item)
             for item in result_list
         ]

src/vws/async_vumark_service.py

@@ -7,6 +7,7 @@
 from beartype import BeartypeConf, beartype
 
 from vws._async_vws_request import async_target_api_request
+from vws._json_utils import json_object, string_field
 from vws.exceptions.base_exceptions import VWSError
 from vws.exceptions.custom_exceptions import ServerError
 from vws.exceptions.vws_exceptions import TooManyRequestsError
@@ -147,8 +148,11 @@ async def generate_vumark_instance(
         if response.status_code == HTTPStatus.OK:
             return response.content
 
-        result_code = json.loads(s=response.text)["result_code"]  # pyrefly: ignore [unknown-variable-type]
+        result_code = string_field(
+            value=json_object(value=response.text),
+            name="result_code",
+        )
         raise VWSError.from_result_code(
-            result_code=result_code,  # pyrefly: ignore [unknown-argument-type]
+            result_code=result_code,
             response=response,
         )

src/vws/async_vws.py

@@ -13,6 +13,7 @@
 from vws._async_vws_request import async_target_api_request
 from vws._image_utils import ImageType as _ImageType
 from vws._image_utils import get_image_data as _get_image_data
+from vws._json_utils import json_object, string_field, string_list_field
 from vws._reco_counts import (
     reco_counts_report_body,
     reco_counts_report_path,
@@ -152,12 +153,15 @@ async def make_request(
         ):  # pragma: no cover
             raise ServerError(response=response)
 
-        result_code = json.loads(s=response.text)["result_code"]  # pyrefly: ignore [unknown-variable-type]
+        result_code = string_field(
+            value=json_object(value=response.text),
+            name="result_code",
+        )
         if result_code == expected_result_code:
             return response
 
         raise VWSError.from_result_code(
-            result_code=result_code,  # pyrefly: ignore [unknown-argument-type]
+            result_code=result_code,
             response=response,
         )
 
@@ -238,7 +242,10 @@ async def add_target(
             content_type="application/json",
         )
 
-        return str(object=json.loads(s=response.text)["target_id"])  # pyrefly: ignore [unknown-argument-type]
+        return string_field(
+            value=json_object(value=response.text),
+            name="target_id",
+        )
 
     async def get_target_record(self, target_id: str) -> TargetStatusAndRecord:
         """Get a given target's target record from the Target
@@ -276,7 +283,7 @@ async def get_target_record(self, target_id: str) -> TargetStatusAndRecord:
             content_type="application/json",
         )
 
-        result_data = json.loads(s=response.text)
+        result_data = json_object(value=response.text)
         return TargetStatusAndRecord.from_response_dict(
             response_dict=result_data,
         )
@@ -373,7 +380,10 @@ async def list_targets(self) -> list[str]:
             content_type="application/json",
         )
 
-        return list(json.loads(s=response.text)["results"])  # pyrefly: ignore [unknown-argument-type]
+        return string_list_field(
+            value=json_object(value=response.text),
+            name="results",
+        )
 
     async def get_target_summary_report(
         self, target_id: str
@@ -413,7 +423,7 @@ async def get_target_summary_report(
             content_type="application/json",
         )
 
-        result_data = dict(json.loads(s=response.text))
+        result_data = json_object(value=response.text)
         return TargetSummaryReport.from_response_dict(
             response_dict=result_data,
         )
@@ -450,7 +460,7 @@ async def get_database_summary_report(
             content_type="application/json",
         )
 
-        response_data = dict(json.loads(s=response.text))
+        response_data = json_object(value=response.text)
         return DatabaseSummaryReport.from_response_dict(
             response_dict=response_data,
         )
@@ -504,7 +514,7 @@ async def request_database_reco_counts_report(
             content_type="application/json",
         )
 
-        response_data = dict(json.loads(s=response.text))
+        response_data = json_object(value=response.text)
         return RecoCountsReportRequest.from_response_dict(
             response_dict=response_data,
         )
@@ -657,8 +667,9 @@ async def get_duplicate_targets(self, target_id: str) -> list[str]:
             content_type="application/json",
         )
 
-        return list(
-            json.loads(s=response.text)["similar_targets"],  # pyrefly: ignore [unknown-argument-type]
+        return string_list_field(
+            value=json_object(value=response.text),
+            name="similar_targets",
         )
 
     async def update_target(

src/vws/exceptions/vws_exceptions.py

@@ -6,11 +6,11 @@
 api#result-codes.
 """
 
-import json
 from urllib.parse import urlparse
 
 from beartype import beartype
 
+from vws._json_utils import json_object, string_field
 from vws.exceptions.base_exceptions import VWSError
 
 
@@ -147,8 +147,8 @@ def target_name(self) -> str:
         if not isinstance(response_body, str | bytes):  # pragma: no cover
             msg = "A target-name error response must have a request body."
             raise TypeError(msg)
-        request_json = json.loads(s=response_body)
-        return str(object=request_json["name"])  # pyrefly: ignore [unknown-argument-type]
+        request_json = json_object(value=response_body)
+        return string_field(value=request_json, name="name")
 
 
 @beartype

src/vws/query.py

@@ -10,6 +10,7 @@
 
 from vws._image_utils import ImageType as _ImageType
 from vws._image_utils import get_image_data as _get_image_data
+from vws._json_utils import json_object, object_list_field, string_field
 from vws.exceptions.base_exceptions import CloudRecoError
 from vws.exceptions.cloud_reco_exceptions import (
     AuthenticationFailureError,
@@ -171,13 +172,16 @@ def query(
             raise CloudRecoError(response=response)
 
         try:
-            response_body = json.loads(s=response.text)
+            response_body = json_object(value=response.text)
         except json.JSONDecodeError as exc:
             if response.status_code >= HTTPStatus.BAD_REQUEST:
                 raise CloudRecoError(response=response) from exc
             raise
 
-        result_code = response_body["result_code"]  # pyrefly: ignore [unknown-variable-type]
+        result_code = string_field(
+            value=response_body,
+            name="result_code",
+        )
         if result_code != "Success":
             exception = {
                 "AuthenticationFailure": AuthenticationFailureError,
@@ -187,8 +191,11 @@ def query(
             }[result_code]
             raise exception(response=response)
 
-        result_list = list(response_body["results"])  # pyrefly: ignore [unknown-argument-type]
+        result_list = object_list_field(
+            value=response_body,
+            name="results",
+        )
         return [
-            QueryResult.from_response_dict(response_dict=item)  # pyrefly: ignore [unknown-argument-type]
+            QueryResult.from_response_dict(response_dict=item)
             for item in result_list
         ]

src/vws/vumark_service.py

@@ -5,6 +5,7 @@
 
 from beartype import BeartypeConf, beartype
 
+from vws._json_utils import json_object, string_field
 from vws._vws_request import target_api_request
 from vws.exceptions.base_exceptions import VWSError
 from vws.exceptions.custom_exceptions import ServerError
@@ -130,8 +131,11 @@ def generate_vumark_instance(
         if response.status_code == HTTPStatus.OK:
             return response.content
 
-        result_code = json.loads(s=response.text)["result_code"]  # pyrefly: ignore [unknown-variable-type]
+        result_code = string_field(
+            value=json_object(value=response.text),
+            name="result_code",
+        )
         raise VWSError.from_result_code(
-            result_code=result_code,  # pyrefly: ignore [unknown-argument-type]
+            result_code=result_code,
             response=response,
         )