Refactor request handling with library-agnostic RequestData

src/mock_vws/_mock_common.py

@@ -1,13 +1,30 @@
 """Common utilities for creating mock routes."""
 
 import json
-from collections.abc import Iterable
+from collections.abc import Iterable, Mapping
 from dataclasses import dataclass
 from typing import Any
 
 from beartype import beartype
 
 
+@dataclass(frozen=True)
+class RequestData:
+    """A library-agnostic representation of an HTTP request.
+
+    Args:
+        method: The HTTP method of the request.
+        path: The path of the request.
+        headers: The headers sent with the request.
+        body: The body of the request.
+    """
+
+    method: str
+    path: str
+    headers: Mapping[str, str]
+    body: bytes
+
+
 @dataclass(frozen=True)
 class Route:
     """A representation of a VWS route.

src/mock_vws/_requests_mock_server/decorators.py

@@ -12,6 +12,7 @@
 from requests import PreparedRequest
 from responses import RequestsMock
 
+from mock_vws._mock_common import RequestData
 from mock_vws.database import CloudDatabase, VuMarkDatabase
 from mock_vws.image_matchers import (
     ImageMatcher,
@@ -27,7 +28,8 @@
 from .mock_web_services_api import MockVuforiaWebServicesAPI
 
 _ResponseType = tuple[int, Mapping[str, str], str | bytes]
-_Callback = Callable[[PreparedRequest], _ResponseType]
+_MockCallback = Callable[[RequestData], _ResponseType]
+_ResponsesCallback = Callable[[PreparedRequest], _ResponseType]
 
 _STRUCTURAL_SIMILARITY_MATCHER = StructuralSimilarityMatcher()
 _BRISQUE_TRACKING_RATER = BrisqueTargetTrackingRater()
@@ -156,10 +158,10 @@ def add_vumark_database(self, vumark_database: VuMarkDatabase) -> None:
 
     @staticmethod
     def _wrap_callback(
-        callback: _Callback,
+        callback: _MockCallback,
         delay_seconds: float,
         sleep_fn: Callable[[float], None],
-    ) -> _Callback:
+    ) -> _ResponsesCallback:
         """Wrap a callback to add a response delay."""
 
         def wrapped(
@@ -186,7 +188,21 @@ def wrapped(
                 sleep_fn(effective)
                 raise requests.exceptions.Timeout
 
-            result = callback(request)
+            raw_body = request.body
+            if raw_body is None:
+                body_bytes = b""
+            elif isinstance(raw_body, str):
+                body_bytes = raw_body.encode(encoding="utf-8")
+            else:
+                body_bytes = raw_body
+
+            request_data = RequestData(
+                method=request.method or "",
+                path=request.path_url,
+                headers=dict(request.headers),
+                body=body_bytes,
+            )
+            result = callback(request_data)
             sleep_fn(delay_seconds)
             return result
 

src/mock_vws/_requests_mock_server/mock_web_query_api.py

@@ -10,9 +10,8 @@
 from typing import ParamSpec, Protocol, runtime_checkable
 
 from beartype import beartype
-from requests.models import PreparedRequest
 
-from mock_vws._mock_common import Route
+from mock_vws._mock_common import RequestData, Route
 from mock_vws._query_tools import (
     get_query_match_response_text,
 )
@@ -78,21 +77,9 @@ def decorator(
     return decorator
 
 
-@beartype
-def _body_bytes(request: PreparedRequest) -> bytes:
-    """Return the body of a request as bytes."""
-    if request.body is None or isinstance(request.body, str):
-        return b""
-
-    return request.body
-
-
 @beartype
 class MockVuforiaWebQueryAPI:
-    """A fake implementation of the Vuforia Web Query API.
-
-    This implementation is tied to the implementation of ``responses``.
-    """
+    """A fake implementation of the Vuforia Web Query API."""
 
     def __init__(
         self,
@@ -114,24 +101,24 @@ def __init__(
         self._query_match_checker = query_match_checker
 
     @route(path_pattern="/v1/query", http_methods={HTTPMethod.POST})
-    def query(self, request: PreparedRequest) -> _ResponseType:
+    def query(self, request: RequestData) -> _ResponseType:
         """Perform an image recognition query."""
         try:
             run_query_validators(
-                request_path=request.path_url,
+                request_path=request.path,
                 request_headers=request.headers,
-                request_body=_body_bytes(request=request),
-                request_method=request.method or "",
+                request_body=request.body,
+                request_method=request.method,
                 databases=self._target_manager.cloud_databases,
             )
         except ValidatorError as exc:
             return exc.status_code, exc.headers, exc.response_text
 
         response_text = get_query_match_response_text(
             request_headers=request.headers,
-            request_body=_body_bytes(request=request),
-            request_method=request.method or "",
-            request_path=request.path_url,
+            request_body=request.body,
+            request_method=request.method,
+            request_path=request.path,
             databases=self._target_manager.cloud_databases,
             query_match_checker=self._query_match_checker,
         )