From d5a0c7131326fd68d13cef47b7b8c48ca53553f9 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Wed, 9 Sep 2026 17:03:28 +0100 Subject: [PATCH 1/3] Type Model Target request payloads --- src/vws/_model_targets.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/src/vws/_model_targets.py b/src/vws/_model_targets.py index 8a8a55c4d..9588aaf6a 100644 --- a/src/vws/_model_targets.py +++ b/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 @@ -169,7 +168,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) -> dict[str, object]: """Get the request representation of a guide view. Args: @@ -178,7 +177,7 @@ 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: dict[str, object] = { "name": view.name, "guideViewPosition": { "rotation": list(view.guide_view_position.rotation), @@ -192,7 +191,7 @@ def _view_dict(*, view: ModelTargetView) -> dict[str, Any]: # pyrefly: ignore [ @beartype(conf=BeartypeConf(is_pep484_tower=True)) -def _model_dict(*, model: ModelTargetModel) -> dict[str, Any]: # pyrefly: ignore [explicit-any] +def _model_dict(*, model: ModelTargetModel) -> dict[str, object]: """Get the request representation of a model. Args: @@ -201,7 +200,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: dict[str, object] = {"name": model.name} optional_values: dict[str, str | None] = { "automaticColoring": model.automatic_coloring, "cadDataBlob": model.cad_data_blob, From 96e74d2baa139768f99201b8a4c54996587c3e64 Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Wed, 9 Sep 2026 17:45:18 +0100 Subject: [PATCH 2/3] Type Model Target payloads as JSON --- src/vws/_model_targets.py | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/src/vws/_model_targets.py b/src/vws/_model_targets.py index 9588aaf6a..ea434cd73 100644 --- a/src/vws/_model_targets.py +++ b/src/vws/_model_targets.py @@ -25,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" @@ -168,7 +173,7 @@ def dataset_download_path( @beartype(conf=BeartypeConf(is_pep484_tower=True)) -def _view_dict(*, view: ModelTargetView) -> dict[str, object]: +def _view_dict(*, view: ModelTargetView) -> _JSONObject: """Get the request representation of a guide view. Args: @@ -177,7 +182,7 @@ def _view_dict(*, view: ModelTargetView) -> dict[str, object]: Returns: The guide view, as it is sent to Vuforia. """ - view_dict: dict[str, object] = { + view_dict: _JSONObject = { "name": view.name, "guideViewPosition": { "rotation": list(view.guide_view_position.rotation), @@ -185,13 +190,14 @@ def _view_dict(*, view: ModelTargetView) -> dict[str, object]: }, } 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, object]: +def _model_dict(*, model: ModelTargetModel) -> _JSONObject: """Get the request representation of a model. Args: @@ -200,7 +206,7 @@ def _model_dict(*, model: ModelTargetModel) -> dict[str, object]: Returns: The model, as it is sent to Vuforia. """ - model_dict: dict[str, object] = {"name": model.name} + model_dict: _JSONObject = {"name": model.name} optional_values: dict[str, str | None] = { "automaticColoring": model.automatic_coloring, "cadDataBlob": model.cad_data_blob, From 66cd8699153de8dfc9daf97eef9e988fff781c0a Mon Sep 17 00:00:00 2001 From: Adam Dangoor Date: Thu, 10 Sep 2026 09:01:53 +0100 Subject: [PATCH 3/3] Consolidate Model Target JSON typing --- src/vws/_json_utils.py | 33 +++++--- src/vws/exceptions/model_target_exceptions.py | 65 +++++---------- src/vws/reports.py | 82 +++++++++++-------- 3 files changed, 91 insertions(+), 89 deletions(-) diff --git a/src/vws/_json_utils.py b/src/vws/_json_utils.py index a5784d7d1..aa60dbc8d 100644 --- a/src/vws/_json_utils.py +++ b/src/vws/_json_utils.py @@ -3,18 +3,24 @@ 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." @@ -22,12 +28,19 @@ def _validated_object(*, value: object) -> dict[str, object]: 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): diff --git a/src/vws/exceptions/model_target_exceptions.py b/src/vws/exceptions/model_target_exceptions.py index 0031d35b3..c810b55de 100644 --- a/src/vws/exceptions/model_target_exceptions.py +++ b/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") ] diff --git a/src/vws/reports.py b/src/vws/reports.py index 431d6a644..86c9a516d 100644 --- a/src/vws/reports.py +++ b/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), )