diff --git a/src/vws/_async_vws_request.py b/src/vws/_async_vws_request.py index 4ebd50c8..169c3591 100644 --- a/src/vws/_async_vws_request.py +++ b/src/vws/_async_vws_request.py @@ -3,8 +3,8 @@ """ from beartype import BeartypeConf, beartype -from vws_auth_tools import authorization_header, rfc_1123_date +from vws._vws_request import build_vws_request_args from vws.response import Response from vws.transports import AsyncTransport @@ -47,27 +47,17 @@ async def async_target_api_request( Returns: The response to the request. """ - date_string = rfc_1123_date() - - signature_string = authorization_header( - access_key=server_access_key, - secret_key=server_secret_key, - method=method, - content=data, + url, headers = build_vws_request_args( content_type=content_type, - date=date_string, + server_access_key=server_access_key, + server_secret_key=server_secret_key, + method=method, + data=data, request_path=request_path, + base_vws_url=base_vws_url, + extra_headers=extra_headers, ) - headers = { - "Authorization": signature_string, - "Date": date_string, - "Content-Type": content_type, - **extra_headers, - } - - url = base_vws_url.rstrip("/") + request_path - return await transport( method=method, url=url, diff --git a/src/vws/_query_common.py b/src/vws/_query_common.py new file mode 100644 index 00000000..4a045aad --- /dev/null +++ b/src/vws/_query_common.py @@ -0,0 +1,26 @@ +"""Shared helpers for CloudReco query implementations.""" + +from typing import NoReturn + +from vws.exceptions.cloud_reco_exceptions import ( + AuthenticationFailureError, + BadImageError, + InactiveProjectError, + RequestTimeTooSkewedError, +) +from vws.response import Response + + +def raise_for_cloud_reco_result_code( + result_code: str, response: Response +) -> NoReturn: + """Raise the appropriate cloud reco exception for the given result + code. + """ + exception = { + "AuthenticationFailure": AuthenticationFailureError, + "BadImage": BadImageError, + "InactiveProject": InactiveProjectError, + "RequestTimeTooSkewed": RequestTimeTooSkewedError, + }[result_code] + raise exception(response=response) diff --git a/src/vws/_vws_common.py b/src/vws/_vws_common.py new file mode 100644 index 00000000..2f3553b8 --- /dev/null +++ b/src/vws/_vws_common.py @@ -0,0 +1,133 @@ +"""Shared helpers for VWS client implementations.""" + +import base64 +import json +from typing import NoReturn + +from vws._image_utils import ImageType, get_image_data +from vws.exceptions.vws_exceptions import ( + AuthenticationFailureError, + BadImageError, + BadRequestError, + DateRangeError, + FailError, + ImageTooLargeError, + InvalidAcceptHeaderError, + InvalidInstanceIdError, + InvalidTargetTypeError, + MetadataTooLargeError, + ProjectHasNoAPIAccessError, + ProjectInactiveError, + ProjectSuspendedError, + RequestQuotaReachedError, + RequestTimeTooSkewedError, + TargetNameExistError, + TargetQuotaReachedError, + TargetStatusNotSuccessError, + TargetStatusProcessingError, + UnknownTargetError, +) +from vws.response import Response + + +def raise_for_vws_result_code( + result_code: str, response: Response +) -> NoReturn: + """Raise the appropriate VWS exception for the given result code.""" + exception = { + "AuthenticationFailure": AuthenticationFailureError, + "BadImage": BadImageError, + "BadRequest": BadRequestError, + "DateRangeError": DateRangeError, + "Fail": FailError, + "ImageTooLarge": ImageTooLargeError, + "MetadataTooLarge": MetadataTooLargeError, + "ProjectHasNoAPIAccess": ProjectHasNoAPIAccessError, + "ProjectInactive": ProjectInactiveError, + "ProjectSuspended": ProjectSuspendedError, + "RequestQuotaReached": RequestQuotaReachedError, + "RequestTimeTooSkewed": RequestTimeTooSkewedError, + "TargetNameExist": TargetNameExistError, + "TargetQuotaReached": TargetQuotaReachedError, + "TargetStatusNotSuccess": TargetStatusNotSuccessError, + "TargetStatusProcessing": TargetStatusProcessingError, + "UnknownTarget": UnknownTargetError, + }[result_code] + raise exception(response=response) + + +def raise_for_vumark_result_code( + result_code: str, response: Response +) -> NoReturn: + """Raise the appropriate VuMark exception for the given result + code. + """ + exception = { + "AuthenticationFailure": AuthenticationFailureError, + "BadRequest": BadRequestError, + "DateRangeError": DateRangeError, + "Fail": FailError, + "InvalidAcceptHeader": InvalidAcceptHeaderError, + "InvalidInstanceId": InvalidInstanceIdError, + "InvalidTargetType": InvalidTargetTypeError, + "RequestTimeTooSkewed": RequestTimeTooSkewedError, + "TargetStatusNotSuccess": TargetStatusNotSuccessError, + "UnknownTarget": UnknownTargetError, + }[result_code] + raise exception(response=response) + + +def build_add_target_content( + *, + name: str, + width: float, + image: ImageType, + active_flag: bool, + application_metadata: str | None, +) -> bytes: + """Build the request body for an add_target request.""" + image_data = get_image_data(image=image) + image_data_encoded = base64.b64encode(s=image_data).decode( + encoding="ascii", + ) + data = { + "name": name, + "width": width, + "image": image_data_encoded, + "active_flag": active_flag, + "application_metadata": application_metadata, + } + return json.dumps(obj=data).encode(encoding="utf-8") + + +def build_update_target_content( + *, + name: str | None, + width: float | None, + image: ImageType | None, + active_flag: bool | None, + application_metadata: str | None, +) -> bytes: + """Build the request body for an update_target request.""" + data: dict[str, str | bool | float | int] = {} + + if name is not None: + data["name"] = name + + if width is not None: + data["width"] = width + + if image is not None: + image_data = get_image_data(image=image) + image_data_encoded = base64.b64encode(s=image_data).decode( + encoding="ascii", + ) + data["image"] = image_data_encoded + + if active_flag is not None: + data["active_flag"] = active_flag + + if application_metadata is not None: + data["application_metadata"] = application_metadata + + return json.dumps(obj=data).encode(encoding="utf-8") diff --git a/src/vws/_vws_request.py b/src/vws/_vws_request.py index 8e08cf86..6d0f0486 100644 --- a/src/vws/_vws_request.py +++ b/src/vws/_vws_request.py @@ -9,6 +9,45 @@ from vws.transports import Transport +def build_vws_request_args( + *, + content_type: str, + server_access_key: str, + server_secret_key: str, + method: str, + data: bytes, + request_path: str, + base_vws_url: str, + extra_headers: dict[str, str], +) -> tuple[str, dict[str, str]]: + """Build the URL and headers for a Vuforia Target API request. + + Returns: + A tuple of (url, headers). + """ + date_string = rfc_1123_date() + + signature_string = authorization_header( + access_key=server_access_key, + secret_key=server_secret_key, + method=method, + content=data, + content_type=content_type, + date=date_string, + request_path=request_path, + ) + + headers = { + "Authorization": signature_string, + "Date": date_string, + "Content-Type": content_type, + **extra_headers, + } + + url = base_vws_url.rstrip("/") + request_path + return url, headers + + @beartype(conf=BeartypeConf(is_pep484_tower=True)) def target_api_request( *, @@ -46,27 +85,17 @@ def target_api_request( Returns: The response to the request. """ - date_string = rfc_1123_date() - - signature_string = authorization_header( - access_key=server_access_key, - secret_key=server_secret_key, - method=method, - content=data, + url, headers = build_vws_request_args( content_type=content_type, - date=date_string, + server_access_key=server_access_key, + server_secret_key=server_secret_key, + method=method, + data=data, request_path=request_path, + base_vws_url=base_vws_url, + extra_headers=extra_headers, ) - headers = { - "Authorization": signature_string, - "Date": date_string, - "Content-Type": content_type, - **extra_headers, - } - - url = base_vws_url.rstrip("/") + request_path - return transport( method=method, url=url, diff --git a/src/vws/async_query.py b/src/vws/async_query.py index f1df2ac7..a10db9c5 100644 --- a/src/vws/async_query.py +++ b/src/vws/async_query.py @@ -12,13 +12,8 @@ from vws._image_utils import ImageType as _ImageType from vws._image_utils import get_image_data as _get_image_data -from vws.exceptions.cloud_reco_exceptions import ( - AuthenticationFailureError, - BadImageError, - InactiveProjectError, - MaxNumResultsOutOfRangeError, - RequestTimeTooSkewedError, -) +from vws._query_common import raise_for_cloud_reco_result_code +from vws.exceptions.cloud_reco_exceptions import MaxNumResultsOutOfRangeError from vws.exceptions.custom_exceptions import ( RequestEntityTooLargeError, ServerError, @@ -188,15 +183,13 @@ async def query( result_code = json.loads(s=response.text)["result_code"] if result_code != "Success": - exception = { - "AuthenticationFailure": (AuthenticationFailureError), - "BadImage": BadImageError, - "InactiveProject": InactiveProjectError, - "RequestTimeTooSkewed": (RequestTimeTooSkewedError), - }[result_code] - raise exception(response=response) - - result_list = list( - json.loads(s=response.text)["results"], - ) - return [QueryResult.from_response_dict(item) for item in result_list] # type: ignore[misc] + raise_for_cloud_reco_result_code( + result_code=result_code, + response=response, + ) + + result_items = list(json.loads(s=response.text)["results"]) + return [ + QueryResult.from_response_dict(response_dict=result_item) + for result_item in result_items + ] diff --git a/src/vws/async_vumark_service.py b/src/vws/async_vumark_service.py index c1e189ee..2ddb6a91 100644 --- a/src/vws/async_vumark_service.py +++ b/src/vws/async_vumark_service.py @@ -7,20 +7,9 @@ from beartype import BeartypeConf, beartype from vws._async_vws_request import async_target_api_request +from vws._vws_common import raise_for_vumark_result_code from vws.exceptions.custom_exceptions import ServerError -from vws.exceptions.vws_exceptions import ( - AuthenticationFailureError, - BadRequestError, - DateRangeError, - FailError, - InvalidAcceptHeaderError, - InvalidInstanceIdError, - InvalidTargetTypeError, - RequestTimeTooSkewedError, - TargetStatusNotSuccessError, - TooManyRequestsError, - UnknownTargetError, -) +from vws.exceptions.vws_exceptions import TooManyRequestsError from vws.transports import AsyncHTTPXTransport, AsyncTransport from vws.vumark_accept import VuMarkAccept @@ -149,22 +138,10 @@ async def generate_vumark_instance( ): # pragma: no cover raise ServerError(response=response) - if response.status_code == HTTPStatus.OK: - return response.content - - result_code = json.loads(s=response.text)["result_code"] - - exception = { - "AuthenticationFailure": (AuthenticationFailureError), - "BadRequest": BadRequestError, - "DateRangeError": DateRangeError, - "Fail": FailError, - "InvalidAcceptHeader": InvalidAcceptHeaderError, - "InvalidInstanceId": InvalidInstanceIdError, - "InvalidTargetType": InvalidTargetTypeError, - "RequestTimeTooSkewed": RequestTimeTooSkewedError, - "TargetStatusNotSuccess": (TargetStatusNotSuccessError), - "UnknownTarget": UnknownTargetError, - }[result_code] - - raise exception(response=response) + if response.status_code != HTTPStatus.OK: + result_code = json.loads(s=response.text)["result_code"] + raise_for_vumark_result_code( + result_code=result_code, + response=response, + ) + return response.content diff --git a/src/vws/async_vws.py b/src/vws/async_vws.py index b81c5e15..21ccdec3 100644 --- a/src/vws/async_vws.py +++ b/src/vws/async_vws.py @@ -1,7 +1,6 @@ """Async tools for interacting with Vuforia APIs.""" import asyncio -import base64 import json from http import HTTPMethod, HTTPStatus from typing import Self @@ -10,31 +9,16 @@ 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._vws_common import ( + build_add_target_content, + build_update_target_content, + raise_for_vws_result_code, +) from vws.exceptions.custom_exceptions import ( ServerError, TargetProcessingTimeoutError, ) -from vws.exceptions.vws_exceptions import ( - AuthenticationFailureError, - BadImageError, - BadRequestError, - DateRangeError, - FailError, - ImageTooLargeError, - MetadataTooLargeError, - ProjectHasNoAPIAccessError, - ProjectInactiveError, - ProjectSuspendedError, - RequestQuotaReachedError, - RequestTimeTooSkewedError, - TargetNameExistError, - TargetQuotaReachedError, - TargetStatusNotSuccessError, - TargetStatusProcessingError, - TooManyRequestsError, - UnknownTargetError, -) +from vws.exceptions.vws_exceptions import TooManyRequestsError from vws.reports import ( DatabaseSummaryReport, TargetStatusAndRecord, @@ -155,30 +139,12 @@ async def make_request( result_code = json.loads(s=response.text)["result_code"] - if result_code == expected_result_code: - return response - - exception = { - "AuthenticationFailure": AuthenticationFailureError, - "BadImage": BadImageError, - "BadRequest": BadRequestError, - "DateRangeError": DateRangeError, - "Fail": FailError, - "ImageTooLarge": ImageTooLargeError, - "MetadataTooLarge": MetadataTooLargeError, - "ProjectHasNoAPIAccess": ProjectHasNoAPIAccessError, - "ProjectInactive": ProjectInactiveError, - "ProjectSuspended": ProjectSuspendedError, - "RequestQuotaReached": RequestQuotaReachedError, - "RequestTimeTooSkewed": RequestTimeTooSkewedError, - "TargetNameExist": TargetNameExistError, - "TargetQuotaReached": TargetQuotaReachedError, - "TargetStatusNotSuccess": TargetStatusNotSuccessError, - "TargetStatusProcessing": TargetStatusProcessingError, - "UnknownTarget": UnknownTargetError, - }[result_code] - - raise exception(response=response) + if result_code != expected_result_code: + raise_for_vws_result_code( + result_code=result_code, + response=response, + ) + return response async def add_target( self, @@ -234,21 +200,14 @@ async def add_target( ~vws.exceptions.vws_exceptions.TooManyRequestsError: Vuforia is rate limiting access. """ - image_data = _get_image_data(image=image) - image_data_encoded = base64.b64encode(s=image_data).decode( - encoding="ascii", + content = build_add_target_content( + name=name, + width=width, + image=image, + active_flag=active_flag, + application_metadata=application_metadata, ) - data = { - "name": name, - "width": width, - "image": image_data_encoded, - "active_flag": active_flag, - "application_metadata": application_metadata, - } - - content = json.dumps(obj=data).encode(encoding="utf-8") - response = await self.make_request( method=HTTPMethod.POST, data=content, @@ -295,8 +254,9 @@ async def get_target_record(self, target_id: str) -> TargetStatusAndRecord: content_type="application/json", ) - result_data = json.loads(s=response.text) - return TargetStatusAndRecord.from_response_dict(result_data) # type: ignore[misc] + return TargetStatusAndRecord.from_response_dict( + response_dict=dict(json.loads(s=response.text)), + ) async def wait_for_target_processed( self, @@ -430,8 +390,9 @@ async def get_target_summary_report( content_type="application/json", ) - result_data = dict(json.loads(s=response.text)) - return TargetSummaryReport.from_response_dict(result_data) # type: ignore[misc] + return TargetSummaryReport.from_response_dict( + response_dict=dict(json.loads(s=response.text)), + ) async def get_database_summary_report( self, @@ -465,8 +426,9 @@ async def get_database_summary_report( content_type="application/json", ) - response_data = dict(json.loads(s=response.text)) - return DatabaseSummaryReport.from_response_dict(response_data) # type: ignore[misc] + return DatabaseSummaryReport.from_response_dict( + response_dict=dict(json.loads(s=response.text)), + ) async def delete_target(self, target_id: str) -> None: """Delete a given target. @@ -602,28 +564,13 @@ async def update_target( ~vws.exceptions.vws_exceptions.TooManyRequestsError: Vuforia is rate limiting access. """ - data: dict[str, str | bool | float | int] = {} - - if name is not None: - data["name"] = name - - if width is not None: - data["width"] = width - - if image is not None: - image_data = _get_image_data(image=image) - image_data_encoded = base64.b64encode( - s=image_data, - ).decode(encoding="ascii") - data["image"] = image_data_encoded - - if active_flag is not None: - data["active_flag"] = active_flag - - if application_metadata is not None: - data["application_metadata"] = application_metadata - - content = json.dumps(obj=data).encode(encoding="utf-8") + content = build_update_target_content( + name=name, + width=width, + image=image, + active_flag=active_flag, + application_metadata=application_metadata, + ) await self.make_request( method=HTTPMethod.PUT, diff --git a/src/vws/query.py b/src/vws/query.py index a4810b73..afd5e89d 100644 --- a/src/vws/query.py +++ b/src/vws/query.py @@ -10,13 +10,8 @@ from vws._image_utils import ImageType as _ImageType from vws._image_utils import get_image_data as _get_image_data -from vws.exceptions.cloud_reco_exceptions import ( - AuthenticationFailureError, - BadImageError, - InactiveProjectError, - MaxNumResultsOutOfRangeError, - RequestTimeTooSkewedError, -) +from vws._query_common import raise_for_cloud_reco_result_code +from vws.exceptions.cloud_reco_exceptions import MaxNumResultsOutOfRangeError from vws.exceptions.custom_exceptions import ( RequestEntityTooLargeError, ServerError, @@ -156,13 +151,13 @@ def query( result_code = json.loads(s=response.text)["result_code"] if result_code != "Success": - exception = { - "AuthenticationFailure": AuthenticationFailureError, - "BadImage": BadImageError, - "InactiveProject": InactiveProjectError, - "RequestTimeTooSkewed": RequestTimeTooSkewedError, - }[result_code] - raise exception(response=response) - - result_list = list(json.loads(s=response.text)["results"]) - return [QueryResult.from_response_dict(item) for item in result_list] # type: ignore[misc] + raise_for_cloud_reco_result_code( + result_code=result_code, + response=response, + ) + + result_items = list(json.loads(s=response.text)["results"]) + return [ + QueryResult.from_response_dict(response_dict=result_item) + for result_item in result_items + ] diff --git a/src/vws/vumark_service.py b/src/vws/vumark_service.py index 47d68d94..33e4f993 100644 --- a/src/vws/vumark_service.py +++ b/src/vws/vumark_service.py @@ -5,21 +5,10 @@ from beartype import BeartypeConf, beartype +from vws._vws_common import raise_for_vumark_result_code from vws._vws_request import target_api_request from vws.exceptions.custom_exceptions import ServerError -from vws.exceptions.vws_exceptions import ( - AuthenticationFailureError, - BadRequestError, - DateRangeError, - FailError, - InvalidAcceptHeaderError, - InvalidInstanceIdError, - InvalidTargetTypeError, - RequestTimeTooSkewedError, - TargetStatusNotSuccessError, - TooManyRequestsError, - UnknownTargetError, -) +from vws.exceptions.vws_exceptions import TooManyRequestsError from vws.transports import RequestsTransport, Transport from vws.vumark_accept import VuMarkAccept @@ -130,22 +119,10 @@ def generate_vumark_instance( ): # pragma: no cover raise ServerError(response=response) - if response.status_code == HTTPStatus.OK: - return response.content - - result_code = json.loads(s=response.text)["result_code"] - - exception = { - "AuthenticationFailure": AuthenticationFailureError, - "BadRequest": BadRequestError, - "DateRangeError": DateRangeError, - "Fail": FailError, - "InvalidAcceptHeader": InvalidAcceptHeaderError, - "InvalidInstanceId": InvalidInstanceIdError, - "InvalidTargetType": InvalidTargetTypeError, - "RequestTimeTooSkewed": RequestTimeTooSkewedError, - "TargetStatusNotSuccess": TargetStatusNotSuccessError, - "UnknownTarget": UnknownTargetError, - }[result_code] - - raise exception(response=response) + if response.status_code != HTTPStatus.OK: + result_code = json.loads(s=response.text)["result_code"] + raise_for_vumark_result_code( + result_code=result_code, + response=response, + ) + return response.content diff --git a/src/vws/vws.py b/src/vws/vws.py index 346b7d62..04194f1f 100644 --- a/src/vws/vws.py +++ b/src/vws/vws.py @@ -1,6 +1,5 @@ """Tools for interacting with Vuforia APIs.""" -import base64 import json import time from http import HTTPMethod, HTTPStatus @@ -8,32 +7,17 @@ from beartype import BeartypeConf, beartype from vws._image_utils import ImageType as _ImageType -from vws._image_utils import get_image_data as _get_image_data +from vws._vws_common import ( + build_add_target_content, + build_update_target_content, + raise_for_vws_result_code, +) from vws._vws_request import target_api_request from vws.exceptions.custom_exceptions import ( ServerError, TargetProcessingTimeoutError, ) -from vws.exceptions.vws_exceptions import ( - AuthenticationFailureError, - BadImageError, - BadRequestError, - DateRangeError, - FailError, - ImageTooLargeError, - MetadataTooLargeError, - ProjectHasNoAPIAccessError, - ProjectInactiveError, - ProjectSuspendedError, - RequestQuotaReachedError, - RequestTimeTooSkewedError, - TargetNameExistError, - TargetQuotaReachedError, - TargetStatusNotSuccessError, - TargetStatusProcessingError, - TooManyRequestsError, - UnknownTargetError, -) +from vws.exceptions.vws_exceptions import TooManyRequestsError from vws.reports import ( DatabaseSummaryReport, TargetStatusAndRecord, @@ -140,30 +124,12 @@ def make_request( result_code = json.loads(s=response.text)["result_code"] - if result_code == expected_result_code: - return response - - exception = { - "AuthenticationFailure": AuthenticationFailureError, - "BadImage": BadImageError, - "BadRequest": BadRequestError, - "DateRangeError": DateRangeError, - "Fail": FailError, - "ImageTooLarge": ImageTooLargeError, - "MetadataTooLarge": MetadataTooLargeError, - "ProjectHasNoAPIAccess": ProjectHasNoAPIAccessError, - "ProjectInactive": ProjectInactiveError, - "ProjectSuspended": ProjectSuspendedError, - "RequestQuotaReached": RequestQuotaReachedError, - "RequestTimeTooSkewed": RequestTimeTooSkewedError, - "TargetNameExist": TargetNameExistError, - "TargetQuotaReached": TargetQuotaReachedError, - "TargetStatusNotSuccess": TargetStatusNotSuccessError, - "TargetStatusProcessing": TargetStatusProcessingError, - "UnknownTarget": UnknownTargetError, - }[result_code] - - raise exception(response=response) + if result_code != expected_result_code: + raise_for_vws_result_code( + result_code=result_code, + response=response, + ) + return response def add_target( self, @@ -219,21 +185,14 @@ def add_target( ~vws.exceptions.vws_exceptions.TooManyRequestsError: Vuforia is rate limiting access. """ - image_data = _get_image_data(image=image) - image_data_encoded = base64.b64encode(s=image_data).decode( - encoding="ascii", + content = build_add_target_content( + name=name, + width=width, + image=image, + active_flag=active_flag, + application_metadata=application_metadata, ) - data = { - "name": name, - "width": width, - "image": image_data_encoded, - "active_flag": active_flag, - "application_metadata": application_metadata, - } - - content = json.dumps(obj=data).encode(encoding="utf-8") - response = self.make_request( method=HTTPMethod.POST, data=content, @@ -280,8 +239,9 @@ def get_target_record(self, target_id: str) -> TargetStatusAndRecord: content_type="application/json", ) - result_data = json.loads(s=response.text) - return TargetStatusAndRecord.from_response_dict(result_data) # type: ignore[misc] + return TargetStatusAndRecord.from_response_dict( + response_dict=dict(json.loads(s=response.text)), + ) def wait_for_target_processed( self, @@ -404,8 +364,9 @@ def get_target_summary_report(self, target_id: str) -> TargetSummaryReport: content_type="application/json", ) - result_data = dict(json.loads(s=response.text)) - return TargetSummaryReport.from_response_dict(result_data) # type: ignore[misc] + return TargetSummaryReport.from_response_dict( + response_dict=dict(json.loads(s=response.text)), + ) def get_database_summary_report(self) -> DatabaseSummaryReport: """Get a summary report for the database. @@ -437,8 +398,9 @@ def get_database_summary_report(self) -> DatabaseSummaryReport: content_type="application/json", ) - response_data = dict(json.loads(s=response.text)) - return DatabaseSummaryReport.from_response_dict(response_data) # type: ignore[misc] + return DatabaseSummaryReport.from_response_dict( + response_dict=dict(json.loads(s=response.text)), + ) def delete_target(self, target_id: str) -> None: """Delete a given target. @@ -568,28 +530,13 @@ def update_target( ~vws.exceptions.vws_exceptions.TooManyRequestsError: Vuforia is rate limiting access. """ - data: dict[str, str | bool | float | int] = {} - - if name is not None: - data["name"] = name - - if width is not None: - data["width"] = width - - if image is not None: - image_data = _get_image_data(image=image) - image_data_encoded = base64.b64encode(s=image_data).decode( - encoding="ascii", - ) - data["image"] = image_data_encoded - - if active_flag is not None: - data["active_flag"] = active_flag - - if application_metadata is not None: - data["application_metadata"] = application_metadata - - content = json.dumps(obj=data).encode(encoding="utf-8") + content = build_update_target_content( + name=name, + width=width, + image=image, + active_flag=active_flag, + application_metadata=application_metadata, + ) self.make_request( method=HTTPMethod.PUT,