Extract shared helpers to reduce sync/async code duplication

Move duplicated exception-raising, response-parsing, and request-building
logic into shared modules (_vws_common.py, _query_common.py) so the sync
and async client implementations can share the same code.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: adamtheturtle

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,

src/vws/_query_common.py

@@ -0,0 +1,58 @@
+"""Shared helpers for CloudReco query implementations."""
+
+import datetime
+import json
+from typing import NoReturn
+
+from vws.exceptions.cloud_reco_exceptions import (
+    AuthenticationFailureError,
+    BadImageError,
+    InactiveProjectError,
+    RequestTimeTooSkewedError,
+)
+from vws.reports import QueryResult, TargetData
+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)
+
+
+def parse_query_results(text: str) -> list[QueryResult]:
+    """Parse the results list from a cloud reco query response body."""
+    result: list[QueryResult] = []
+    result_list = list(json.loads(s=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

src/vws/_vws_common.py

@@ -0,0 +1,195 @@
+"""Shared helpers for VWS client implementations."""
+
+import base64
+import json
+from datetime import date
+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.reports import (
+    DatabaseSummaryReport,
+    TargetRecord,
+    TargetStatusAndRecord,
+    TargetStatuses,
+    TargetSummaryReport,
+)
+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 parse_target_record_response(text: str) -> TargetStatusAndRecord:
+    """Parse a get_target_record response body."""
+    result_data = json.loads(s=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,
+    )
+
+
+def parse_target_summary_response(text: str) -> TargetSummaryReport:
+    """Parse a get_target_summary_report response body."""
+    result_data = dict(json.loads(s=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"]),
+    )
+
+
+def parse_database_summary_response(text: str) -> DatabaseSummaryReport:
+    """Parse a get_database_summary_report response body."""
+    response_data = dict(json.loads(s=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"]),
+    )
+
+
+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")

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,