Type and validate API JSON payloads

src/vws/_json_utils.py

@@ -3,31 +3,44 @@
 import json
 from typing import TypeGuard
 
+from beartype.door import TypeHint
 
-def _is_json_object(value: object, /) -> TypeGuard[dict[str, object]]:
+type JSONValue = (
+    bool | int | float | str | list[JSONValue] | dict[str, JSONValue] | None
+)
+
+
+def _is_json_object(value: object, /) -> TypeGuard[dict[str, JSONValue]]:
     """Return whether a decoded JSON value is an object."""
-    return isinstance(value, dict)
+    return TypeHint(hint=dict[str, JSONValue]).is_bearable(obj=value)
 
 
-def _is_object_list(value: object, /) -> TypeGuard[list[object]]:
+def _is_object_list(value: object, /) -> TypeGuard[list[JSONValue]]:
     """Return whether a decoded JSON value is an array."""
-    return isinstance(value, list)
+    return TypeHint(hint=list[JSONValue]).is_bearable(obj=value)
 
 
-def _validated_object(*, value: object) -> dict[str, object]:
+def _validated_object(*, value: object) -> dict[str, JSONValue]:
     """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]:
+def json_object(*, value: str | bytes | bytearray) -> dict[str, JSONValue]:
     """Decode and validate a JSON object."""
     loaded: object = json.loads(s=value)
     return _validated_object(value=loaded)
 
 
+def object_field(
+    *, value: dict[str, JSONValue], name: str
+) -> dict[str, JSONValue]:
+    """Return a required JSON object field."""
+    return _validated_object(value=value[name])
+
+
 def string_value(*, value: object, name: str) -> str:
     """Return a JSON value after validating that it is a string."""
     if not isinstance(value, str):
@@ -36,14 +49,14 @@ def string_value(*, value: object, name: str) -> str:
     return value
 
 
-def string_field(*, value: dict[str, object], name: str) -> str:
+def string_field(*, value: dict[str, JSONValue], 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],
+    value: dict[str, JSONValue],
     name: str,
 ) -> list[str]:
     """Return a required list of strings from a JSON object."""
@@ -58,9 +71,9 @@ def string_list_field(
 
 def object_list_field(
     *,
-    value: dict[str, object],
+    value: dict[str, JSONValue],
     name: str,
-) -> list[dict[str, object]]:
+) -> list[dict[str, JSONValue]]:
     """Return a required list of JSON objects."""
     items = value[name]
     if not _is_object_list(items):

src/vws/_model_targets.py

@@ -4,7 +4,6 @@
 import json
 from collections.abc import Sequence
 from http import HTTPStatus
-from typing import Any
 
 from beartype import BeartypeConf, beartype
 
@@ -26,6 +25,11 @@
 from vws.reports import ModelTargetDatasetStatusReport
 from vws.response import Response
 
+type _JSONValue = (
+    bool | int | float | str | list[_JSONValue] | dict[str, _JSONValue] | None
+)
+type _JSONObject = dict[str, _JSONValue]
+
 OAUTH2_ENDPOINT_PATH = "/oauth2/token"
 OAUTH2_TOKEN_BODY = b"grant_type=client_credentials"
 OAUTH2_MEDIA_TYPE = "application/x-www-form-urlencoded"
@@ -169,7 +173,7 @@ def dataset_download_path(
 
 
 @beartype(conf=BeartypeConf(is_pep484_tower=True))
-def _view_dict(*, view: ModelTargetView) -> dict[str, Any]:  # pyrefly: ignore [explicit-any]
+def _view_dict(*, view: ModelTargetView) -> _JSONObject:
     """Get the request representation of a guide view.
 
     Args:
@@ -178,21 +182,22 @@ def _view_dict(*, view: ModelTargetView) -> dict[str, Any]:  # pyrefly: ignore [
     Returns:
         The guide view, as it is sent to Vuforia.
     """
-    view_dict: dict[str, Any] = {  # pyrefly: ignore [explicit-any]
+    view_dict: _JSONObject = {
         "name": view.name,
         "guideViewPosition": {
             "rotation": list(view.guide_view_position.rotation),
             "translation": list(view.guide_view_position.translation),
         },
     }
     if view.states is not None:
-        view_dict["states"] = list(view.states)
+        states = list[_JSONValue](view.states)
+        view_dict["states"] = states
 
     return view_dict
 
 
 @beartype(conf=BeartypeConf(is_pep484_tower=True))
-def _model_dict(*, model: ModelTargetModel) -> dict[str, Any]:  # pyrefly: ignore [explicit-any]
+def _model_dict(*, model: ModelTargetModel) -> _JSONObject:
     """Get the request representation of a model.
 
     Args:
@@ -201,7 +206,7 @@ def _model_dict(*, model: ModelTargetModel) -> dict[str, Any]:  # pyrefly: ignor
     Returns:
         The model, as it is sent to Vuforia.
     """
-    model_dict: dict[str, Any] = {"name": model.name}  # pyrefly: ignore [explicit-any]
+    model_dict: _JSONObject = {"name": model.name}
     optional_values: dict[str, str | None] = {
         "automaticColoring": model.automatic_coloring,
         "cadDataBlob": model.cad_data_blob,

src/vws/exceptions/model_target_exceptions.py

@@ -5,52 +5,33 @@
 """
 
 import json
-from typing import Any
 
 from beartype import beartype
 
+from vws._json_utils import (
+    JSONValue,
+    json_object,
+    object_field,
+    object_list_field,
+    string_field,
+)
 from vws.reports import ModelTargetGenerationDetail
 from vws.response import Response
 
 
 @beartype
-def _is_json_object(*, value: object) -> bool:
-    """Get whether a decoded JSON value is an object.
-
-    Args:
-        value: A decoded JSON value.
-
-    Returns:
-        Whether the value is a JSON object.
-    """
-    return isinstance(value, dict)
-
-
-@beartype
-def _json_object(*, value: str) -> dict[str, Any]:  # pyrefly: ignore [explicit-any]
-    """Get a JSON object from a string.
-
-    Args:
-        value: A string which may be a JSON object.
-
-    Returns:
-        The JSON object, or an empty dictionary if the string is not a
-        JSON object.
+def _json_object(*, value: str) -> dict[str, JSONValue]:
+    """Return a decoded JSON object, or an empty object for invalid
+    input.
     """
     try:
-        loaded: Any = json.loads(s=value)  # pyrefly: ignore [explicit-any]
-    except json.JSONDecodeError:
+        return json_object(value=value)
+    except json.JSONDecodeError, TypeError:
         return {}
 
-    if not _is_json_object(value=loaded):
-        return {}
-
-    json_object: dict[str, Any] = loaded  # pyrefly: ignore [explicit-any]
-    return json_object
-
 
 @beartype
-def _error_dict(*, response: Response) -> dict[str, Any]:  # pyrefly: ignore [explicit-any]
+def _error_dict(*, response: Response) -> dict[str, JSONValue]:
     """Get the error object of a Model Target Web API error response.
 
     Args:
@@ -62,17 +43,12 @@ def _error_dict(*, response: Response) -> dict[str, Any]:  # pyrefly: ignore [ex
         balancer in front of Vuforia, are not shaped like Model Target
         Web API errors.
     """
-    body = _json_object(value=response.text)
-    if "error" not in body:
-        return {}
-
-    error: Any = body["error"]  # pyrefly: ignore [explicit-any]
-    if not _is_json_object(value=error):
+    try:
+        body = _json_object(value=response.text)
+        return object_field(value=body, name="error")
+    except KeyError, TypeError:
         return {}
 
-    error_dict: dict[str, Any] = error  # pyrefly: ignore [explicit-any]
-    return error_dict
-
 
 @beartype
 class ModelTargetError(Exception):
@@ -122,11 +98,10 @@ def details(self) -> list[ModelTargetGenerationDetail]:
 
         return [
             ModelTargetGenerationDetail(
-                code=detail["code"],  # pyrefly: ignore [unknown-argument-type]
-                # pyrefly: ignore [unknown-argument-type]
-                message=detail["message"],
+                code=string_field(value=detail_object, name="code"),
+                message=string_field(value=detail_object, name="message"),
             )
-            for detail in error["details"]
+            for detail_object in object_list_field(value=error, name="details")
         ]
 
 

src/vws/reports.py

@@ -6,7 +6,7 @@
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from enum import Enum, unique
-from typing import Any, Self, TypeIs
+from typing import Self, TypeIs
 
 from beartype import BeartypeConf, beartype
 from beartype.door import TypeHint
@@ -64,21 +64,25 @@ class DatabaseSummaryReport:
     total_recos: int
 
     @classmethod
-    def from_response_dict(cls, response_dict: dict[str, Any]) -> Self:  # pyrefly: ignore [explicit-any]
+    def from_response_dict(cls, response_dict: Mapping[str, object]) -> Self:
         """Construct from a VWS API response dict."""
         return cls(
-            active_images=int(response_dict["active_images"]),
-            current_month_recos=int(response_dict["current_month_recos"]),
-            failed_images=int(response_dict["failed_images"]),
-            inactive_images=int(response_dict["inactive_images"]),
-            name=response_dict["name"],
-            previous_month_recos=int(response_dict["previous_month_recos"]),
-            processing_images=int(response_dict["processing_images"]),
-            reco_threshold=int(response_dict["reco_threshold"]),
-            request_quota=int(response_dict["request_quota"]),
-            request_usage=int(response_dict["request_usage"]),
-            target_quota=int(response_dict["target_quota"]),
-            total_recos=int(response_dict["total_recos"]),
+            active_images=int(_number(response_dict["active_images"])),
+            current_month_recos=int(
+                _number(response_dict["current_month_recos"])
+            ),
+            failed_images=int(_number(response_dict["failed_images"])),
+            inactive_images=int(_number(response_dict["inactive_images"])),
+            name=_checked(response_dict["name"], str),
+            previous_month_recos=int(
+                _number(response_dict["previous_month_recos"])
+            ),
+            processing_images=int(_number(response_dict["processing_images"])),
+            reco_threshold=int(_number(response_dict["reco_threshold"])),
+            request_quota=int(_number(response_dict["request_quota"])),
+            request_usage=int(_number(response_dict["request_usage"])),
+            target_quota=int(_number(response_dict["target_quota"])),
+            total_recos=int(_number(response_dict["total_recos"])),
         )
 
 
@@ -116,20 +120,26 @@ class TargetSummaryReport:
     previous_month_recos: int
 
     @classmethod
-    def from_response_dict(cls, response_dict: dict[str, Any]) -> Self:  # pyrefly: ignore [explicit-any]
+    def from_response_dict(cls, response_dict: Mapping[str, object]) -> Self:
         """Construct from a VWS API response dict."""
         return cls(
-            status=TargetStatuses(value=response_dict["status"]),
-            database_name=response_dict["database_name"],
-            target_name=response_dict["target_name"],
+            status=TargetStatuses(
+                value=_checked(response_dict["status"], str)
+            ),
+            database_name=_checked(response_dict["database_name"], str),
+            target_name=_checked(response_dict["target_name"], str),
             upload_date=datetime.date.fromisoformat(
-                response_dict["upload_date"]
+                _checked(response_dict["upload_date"], str)
             ),
             active_flag=bool(response_dict["active_flag"]),
-            tracking_rating=int(response_dict["tracking_rating"]),
-            total_recos=int(response_dict["total_recos"]),
-            current_month_recos=int(response_dict["current_month_recos"]),
-            previous_month_recos=int(response_dict["previous_month_recos"]),
+            tracking_rating=int(_number(response_dict["tracking_rating"])),
+            total_recos=int(_number(response_dict["total_recos"])),
+            current_month_recos=int(
+                _number(response_dict["current_month_recos"])
+            ),
+            previous_month_recos=int(
+                _number(response_dict["previous_month_recos"])
+            ),
         )
 
 
@@ -213,17 +223,21 @@ class TargetStatusAndRecord:
     target_record: TargetRecord
 
     @classmethod
-    def from_response_dict(cls, response_dict: dict[str, Any]) -> Self:  # pyrefly: ignore [explicit-any]
+    def from_response_dict(cls, response_dict: Mapping[str, object]) -> Self:
         """Construct from a VWS API response dict."""
-        status = TargetStatuses(value=response_dict["status"])
-        target_record_dict = dict(response_dict["target_record"])
+        status = TargetStatuses(value=_checked(response_dict["status"], str))
+        target_record_dict = _checked(
+            response_dict["target_record"], dict[str, object]
+        )
         target_record = TargetRecord(
-            target_id=target_record_dict["target_id"],
+            target_id=_checked(target_record_dict["target_id"], str),
             active_flag=bool(target_record_dict["active_flag"]),
-            name=target_record_dict["name"],
-            width=float(target_record_dict["width"]),
-            tracking_rating=int(target_record_dict["tracking_rating"]),
-            reco_rating=target_record_dict["reco_rating"],
+            name=_checked(target_record_dict["name"], str),
+            width=float(_number(target_record_dict["width"])),
+            tracking_rating=int(
+                _number(target_record_dict["tracking_rating"])
+            ),
+            reco_rating=_checked(target_record_dict["reco_rating"], str),
         )
         return cls(status=status, target_record=target_record)
 
@@ -246,11 +260,11 @@ class RecoCountsReportRequest:
     """
 
     @classmethod
-    def from_response_dict(cls, response_dict: dict[str, Any]) -> Self:  # pyrefly: ignore [explicit-any]
+    def from_response_dict(cls, response_dict: Mapping[str, object]) -> Self:
         """Construct from a VWS API response dict."""
         return cls(
-            transaction_id=response_dict["transaction_id"],
-            presigned_url=response_dict["presigned_url"],
+            transaction_id=_checked(response_dict["transaction_id"], str),
+            presigned_url=_checked(response_dict["presigned_url"], str),
         )