diff --git a/src/vws/async_query.py b/src/vws/async_query.py index 56981d4e..f1df2ac7 100644 --- a/src/vws/async_query.py +++ b/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] diff --git a/src/vws/async_vws.py b/src/vws/async_vws.py index f0f9005a..b81c5e15 100644 --- a/src/vws/async_vws.py +++ b/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. diff --git a/src/vws/query.py b/src/vws/query.py index 4b5433f3..a4810b73 100644 --- a/src/vws/query.py +++ b/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] diff --git a/src/vws/reports.py b/src/vws/reports.py index e59408aa..726f93b6 100644 --- a/src/vws/reports.py +++ b/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) diff --git a/src/vws/vws.py b/src/vws/vws.py index bed87c93..346b7d62 100644 --- a/src/vws/vws.py +++ b/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.