Merge pull request #2882 from VWS-Python/adamtheturtle/reports-from-response-dict

Add from_response_dict classmethods to report classes

Author: adamtheturtle

src/vws/async_query.py

@@ -2,7 +2,6 @@
 Web APIs.
 """
 
-import datetime
 import json
 from http import HTTPMethod, HTTPStatus
 from typing import Any, Self
@@ -25,7 +24,7 @@
     ServerError,
 )
 from vws.include_target_data import CloudRecoIncludeTargetData
-from vws.reports import QueryResult, TargetData
+from vws.reports import QueryResult
 from vws.transports import AsyncHTTPXTransport, AsyncTransport
 
 
@@ -197,30 +196,7 @@ async def query(
             }[result_code]
             raise exception(response=response)
 
-        result: list[QueryResult] = []
         result_list = list(
             json.loads(s=response.text)["results"],
         )
-        for item in result_list:
-            target_data: TargetData | None = None
-            if "target_data" in item:
-                target_data_dict = item["target_data"]
-                metadata = target_data_dict["application_metadata"]
-                timestamp_string = target_data_dict["target_timestamp"]
-                target_timestamp = datetime.datetime.fromtimestamp(
-                    timestamp=timestamp_string,
-                    tz=datetime.UTC,
-                )
-                target_data = TargetData(
-                    name=target_data_dict["name"],
-                    application_metadata=metadata,
-                    target_timestamp=target_timestamp,
-                )
-
-            query_result = QueryResult(
-                target_id=item["target_id"],
-                target_data=target_data,
-            )
-
-            result.append(query_result)
-        return result
+        return [QueryResult.from_response_dict(item) for item in result_list]  # type: ignore[misc]

src/vws/async_vws.py

@@ -3,7 +3,6 @@
 import asyncio
 import base64
 import json
-from datetime import date
 from http import HTTPMethod, HTTPStatus
 from typing import Self
 
@@ -38,7 +37,6 @@
 )
 from vws.reports import (
     DatabaseSummaryReport,
-    TargetRecord,
     TargetStatusAndRecord,
     TargetStatuses,
     TargetSummaryReport,
@@ -298,20 +296,7 @@ async def get_target_record(self, target_id: str) -> TargetStatusAndRecord:
         )
 
         result_data = json.loads(s=response.text)
-        status = TargetStatuses(value=result_data["status"])
-        target_record_dict = dict(result_data["target_record"])
-        target_record = TargetRecord(
-            target_id=target_record_dict["target_id"],
-            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"],
-        )
-        return TargetStatusAndRecord(
-            status=status,
-            target_record=target_record,
-        )
+        return TargetStatusAndRecord.from_response_dict(result_data)  # type: ignore[misc]
 
     async def wait_for_target_processed(
         self,
@@ -446,27 +431,7 @@ async def get_target_summary_report(
         )
 
         result_data = dict(json.loads(s=response.text))
-        return TargetSummaryReport(
-            status=TargetStatuses(
-                value=result_data["status"],
-            ),
-            database_name=result_data["database_name"],
-            target_name=result_data["target_name"],
-            upload_date=date.fromisoformat(
-                result_data["upload_date"],
-            ),
-            active_flag=bool(result_data["active_flag"]),
-            tracking_rating=int(
-                result_data["tracking_rating"],
-            ),
-            total_recos=int(result_data["total_recos"]),
-            current_month_recos=int(
-                result_data["current_month_recos"],
-            ),
-            previous_month_recos=int(
-                result_data["previous_month_recos"],
-            ),
-        )
+        return TargetSummaryReport.from_response_dict(result_data)  # type: ignore[misc]
 
     async def get_database_summary_report(
         self,
@@ -501,30 +466,7 @@ async def get_database_summary_report(
         )
 
         response_data = dict(json.loads(s=response.text))
-        return DatabaseSummaryReport(
-            active_images=int(response_data["active_images"]),
-            current_month_recos=int(
-                response_data["current_month_recos"],
-            ),
-            failed_images=int(response_data["failed_images"]),
-            inactive_images=int(
-                response_data["inactive_images"],
-            ),
-            name=str(object=response_data["name"]),
-            previous_month_recos=int(
-                response_data["previous_month_recos"],
-            ),
-            processing_images=int(
-                response_data["processing_images"],
-            ),
-            reco_threshold=int(
-                response_data["reco_threshold"],
-            ),
-            request_quota=int(response_data["request_quota"]),
-            request_usage=int(response_data["request_usage"]),
-            target_quota=int(response_data["target_quota"]),
-            total_recos=int(response_data["total_recos"]),
-        )
+        return DatabaseSummaryReport.from_response_dict(response_data)  # type: ignore[misc]
 
     async def delete_target(self, target_id: str) -> None:
         """Delete a given target.

src/vws/query.py

@@ -1,6 +1,5 @@
 """Tools for interacting with the Vuforia Cloud Recognition Web APIs."""
 
-import datetime
 import json
 from http import HTTPMethod, HTTPStatus
 from typing import Any
@@ -23,7 +22,7 @@
     ServerError,
 )
 from vws.include_target_data import CloudRecoIncludeTargetData
-from vws.reports import QueryResult, TargetData
+from vws.reports import QueryResult
 from vws.transports import RequestsTransport, Transport
 
 
@@ -165,28 +164,5 @@ def query(
             }[result_code]
             raise exception(response=response)
 
-        result: list[QueryResult] = []
         result_list = list(json.loads(s=response.text)["results"])
-        for item in result_list:
-            target_data: TargetData | None = None
-            if "target_data" in item:
-                target_data_dict = item["target_data"]
-                metadata = target_data_dict["application_metadata"]
-                timestamp_string = target_data_dict["target_timestamp"]
-                target_timestamp = datetime.datetime.fromtimestamp(
-                    timestamp=timestamp_string,
-                    tz=datetime.UTC,
-                )
-                target_data = TargetData(
-                    name=target_data_dict["name"],
-                    application_metadata=metadata,
-                    target_timestamp=target_timestamp,
-                )
-
-            query_result = QueryResult(
-                target_id=item["target_id"],
-                target_data=target_data,
-            )
-
-            result.append(query_result)
-        return result
+        return [QueryResult.from_response_dict(item) for item in result_list]  # type: ignore[misc]

src/vws/reports.py

@@ -3,6 +3,7 @@
 import datetime
 from dataclasses import dataclass
 from enum import Enum, unique
+from typing import Any, Self
 
 from beartype import BeartypeConf, beartype
 
@@ -29,6 +30,24 @@ class DatabaseSummaryReport:
     target_quota: int
     total_recos: int
 
+    @classmethod
+    def from_response_dict(cls, response_dict: dict[str, Any]) -> 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"]),
+        )
+
 
 @beartype
 @unique
@@ -63,6 +82,23 @@ class TargetSummaryReport:
     current_month_recos: int
     previous_month_recos: int
 
+    @classmethod
+    def from_response_dict(cls, response_dict: dict[str, Any]) -> 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"],
+            upload_date=datetime.date.fromisoformat(
+                response_dict["upload_date"]
+            ),
+            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"]),
+        )
+
 
 @beartype(conf=BeartypeConf(is_pep484_tower=True))
 @dataclass(frozen=True)
@@ -103,6 +139,26 @@ class QueryResult:
     target_id: str
     target_data: TargetData | None
 
+    @classmethod
+    def from_response_dict(cls, response_dict: dict[str, Any]) -> Self:
+        """Construct from a VWS API query result item dict."""
+        target_data: TargetData | None = None
+        if "target_data" in response_dict:
+            target_data_dict = response_dict["target_data"]
+            target_timestamp = datetime.datetime.fromtimestamp(
+                timestamp=target_data_dict["target_timestamp"],
+                tz=datetime.UTC,
+            )
+            target_data = TargetData(
+                name=target_data_dict["name"],
+                application_metadata=target_data_dict["application_metadata"],
+                target_timestamp=target_timestamp,
+            )
+        return cls(
+            target_id=response_dict["target_id"],
+            target_data=target_data,
+        )
+
 
 @beartype
 @dataclass(frozen=True)
@@ -115,3 +171,18 @@ class TargetStatusAndRecord:
 
     status: TargetStatuses
     target_record: TargetRecord
+
+    @classmethod
+    def from_response_dict(cls, response_dict: dict[str, Any]) -> Self:
+        """Construct from a VWS API response dict."""
+        status = TargetStatuses(value=response_dict["status"])
+        target_record_dict = dict(response_dict["target_record"])
+        target_record = TargetRecord(
+            target_id=target_record_dict["target_id"],
+            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"],
+        )
+        return cls(status=status, target_record=target_record)

src/vws/vws.py

@@ -3,7 +3,6 @@
 import base64
 import json
 import time
-from datetime import date
 from http import HTTPMethod, HTTPStatus
 
 from beartype import BeartypeConf, beartype
@@ -37,7 +36,6 @@
 )
 from vws.reports import (
     DatabaseSummaryReport,
-    TargetRecord,
     TargetStatusAndRecord,
     TargetStatuses,
     TargetSummaryReport,
@@ -283,20 +281,7 @@ def get_target_record(self, target_id: str) -> TargetStatusAndRecord:
         )
 
         result_data = json.loads(s=response.text)
-        status = TargetStatuses(value=result_data["status"])
-        target_record_dict = dict(result_data["target_record"])
-        target_record = TargetRecord(
-            target_id=target_record_dict["target_id"],
-            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"],
-        )
-        return TargetStatusAndRecord(
-            status=status,
-            target_record=target_record,
-        )
+        return TargetStatusAndRecord.from_response_dict(result_data)  # type: ignore[misc]
 
     def wait_for_target_processed(
         self,
@@ -420,17 +405,7 @@ def get_target_summary_report(self, target_id: str) -> TargetSummaryReport:
         )
 
         result_data = dict(json.loads(s=response.text))
-        return TargetSummaryReport(
-            status=TargetStatuses(value=result_data["status"]),
-            database_name=result_data["database_name"],
-            target_name=result_data["target_name"],
-            upload_date=date.fromisoformat(result_data["upload_date"]),
-            active_flag=bool(result_data["active_flag"]),
-            tracking_rating=int(result_data["tracking_rating"]),
-            total_recos=int(result_data["total_recos"]),
-            current_month_recos=int(result_data["current_month_recos"]),
-            previous_month_recos=int(result_data["previous_month_recos"]),
-        )
+        return TargetSummaryReport.from_response_dict(result_data)  # type: ignore[misc]
 
     def get_database_summary_report(self) -> DatabaseSummaryReport:
         """Get a summary report for the database.
@@ -463,20 +438,7 @@ def get_database_summary_report(self) -> DatabaseSummaryReport:
         )
 
         response_data = dict(json.loads(s=response.text))
-        return DatabaseSummaryReport(
-            active_images=int(response_data["active_images"]),
-            current_month_recos=int(response_data["current_month_recos"]),
-            failed_images=int(response_data["failed_images"]),
-            inactive_images=int(response_data["inactive_images"]),
-            name=str(object=response_data["name"]),
-            previous_month_recos=int(response_data["previous_month_recos"]),
-            processing_images=int(response_data["processing_images"]),
-            reco_threshold=int(response_data["reco_threshold"]),
-            request_quota=int(response_data["request_quota"]),
-            request_usage=int(response_data["request_usage"]),
-            target_quota=int(response_data["target_quota"]),
-            total_recos=int(response_data["total_recos"]),
-        )
+        return DatabaseSummaryReport.from_response_dict(response_data)  # type: ignore[misc]
 
     def delete_target(self, target_id: str) -> None:
         """Delete a given target.