Extract shared helpers and fix strict-kwargs call sites

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,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)

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")

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,

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
+        ]

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