Add database and target type support with VuMark validation

CHANGELOG.rst

@@ -4,6 +4,10 @@ Changelog
 Next
 ----
 
+- Add ``VuMarkTarget`` class for VuMark template targets, alongside the renamed ``ImageTarget`` class (previously ``Target``).
+  ``ImageTarget`` is for image-based targets and ``VuMarkTarget`` is for VuMark template targets.
+  Both can be stored in a ``VuforiaDatabase``.
+
 2026.02.18.2
 ------------
 

docs/source/mock-api-reference.rst

@@ -29,6 +29,10 @@ API Reference
    :members:
    :undoc-members:
 
+.. autoenum:: mock_vws.database_type.DatabaseType
+   :members:
+   :undoc-members:
+
 .. autoclass:: mock_vws.target.ImageTarget
 
 .. autoclass:: mock_vws.target.VuMarkTarget

pyproject.toml

@@ -452,6 +452,7 @@ ignore_names = [
     # Used in TYPE_CHECKING for type hints
     "CloudDatabaseDict",
     "VuMarkDatabaseDict",
+    "VuMarkTargetDict",
 ]
 # Duplicate some of .gitignore
 exclude = [ ".venv" ]

spelling_private_dict.txt

@@ -3,6 +3,7 @@ MPixel
 MiB
 MissingSchema
 Ubuntu
+VuMark
 admin
 another's
 api

src/mock_vws/_constants.py

@@ -63,6 +63,7 @@ class ResultCodes(Enum):
     INVALID_ACCEPT_HEADER = "InvalidAcceptHeader"
     INVALID_INSTANCE_ID = "InvalidInstanceId"
     BAD_REQUEST = "BadRequest"
+    INVALID_TARGET_TYPE = "InvalidTargetType"
 
 
 @beartype

src/mock_vws/_flask_server/target_manager.py

@@ -13,6 +13,7 @@
 from pydantic_settings import BaseSettings
 
 from mock_vws.database import CloudDatabase, VuMarkDatabase
+from mock_vws.database_type import DatabaseType
 from mock_vws.states import States
 from mock_vws.target import ImageTarget, VuMarkTarget
 from mock_vws.target_manager import TargetManager
@@ -204,8 +205,13 @@ def create_cloud_database() -> Response:
         "state_name",
         random_database.state.name,
     )
+    database_type_name = request_json.get(
+        "database_type_name",
+        random_database.database_type.name,
+    )
 
     state = States[state_name]
+    database_type = DatabaseType[database_type_name]
 
     database = CloudDatabase(
         server_access_key=server_access_key,
@@ -214,6 +220,7 @@ def create_cloud_database() -> Response:
         client_secret_key=client_secret_key,
         database_name=database_name,
         state=state,
+        database_type=database_type,
     )
     try:
         TARGET_MANAGER.add_cloud_database(cloud_database=database)
@@ -283,11 +290,10 @@ def create_target(database_name: str) -> Response:
         if database.database_name == database_name
     )
     request_json = json.loads(s=request.data)
-    image_base64 = request_json["image_base64"]
-    image_bytes = base64.b64decode(s=image_base64)
     settings = TargetManagerSettings.model_validate(obj={})
-    target_tracking_rater = settings.target_rater.to_target_rater()
 
+    image_bytes = base64.b64decode(s=request_json["image_base64"])
+    target_tracking_rater = settings.target_rater.to_target_rater()
     target = ImageTarget(
         name=request_json["name"],
         width=request_json["width"],
@@ -343,7 +349,7 @@ def delete_target(database_name: str, target_id: str) -> Response:
     target = database.get_target(target_id=target_id)
     now = datetime.datetime.now(tz=target.upload_date.tzinfo)
     # See https://github.com/facebook/pyrefly/issues/1897
-    new_target = copy.replace(
+    new_target: ImageTarget = copy.replace(
         target,  # pyrefly: ignore[bad-argument-type]
         delete_date=now,
     )
@@ -369,24 +375,22 @@ def update_target(database_name: str, target_id: str) -> Response:
     target = database.get_target(target_id=target_id)
 
     request_json = json.loads(s=request.data)
-    width = request_json.get("width", target.width)
     name = request_json.get("name", target.name)
     active_flag = request_json.get("active_flag", target.active_flag)
+
+    gmt = ZoneInfo(key="GMT")
+    last_modified_date = datetime.datetime.now(tz=gmt)
+
+    width = request_json.get("width", target.width)
     application_metadata = request_json.get(
         "application_metadata",
         target.application_metadata,
     )
-
     image_value = target.image_value
-    request_json = json.loads(s=request.data)
     if "image" in request_json:
         image_value = base64.b64decode(s=request_json["image"])
-
-    gmt = ZoneInfo(key="GMT")
-    last_modified_date = datetime.datetime.now(tz=gmt)
-
     # See https://github.com/facebook/pyrefly/issues/1897
-    new_target = copy.replace(
+    new_target: ImageTarget = copy.replace(
         target,  # pyrefly: ignore[bad-argument-type]
         name=name,
         width=width,

src/mock_vws/_flask_server/vws.py

@@ -32,6 +32,7 @@
     FailError,
     InvalidAcceptHeaderError,
     InvalidInstanceIdError,
+    InvalidTargetTypeError,
     TargetStatusNotSuccessError,
     TargetStatusProcessingError,
     ValidatorError,
@@ -278,13 +279,16 @@ def get_target(target_id: str) -> Response:
         target for target in database.targets if target.target_id == target_id
     )
 
+    width = target.width
+    tracking_rating = target.tracking_rating
+    reco_rating = target.reco_rating
     target_record = {
         "target_id": target.target_id,
         "active_flag": target.active_flag,
         "name": target.name,
-        "width": target.width,
-        "tracking_rating": target.tracking_rating,
-        "reco_rating": target.reco_rating,
+        "width": width,
+        "tracking_rating": tracking_rating,
+        "reco_rating": reco_rating,
     }
 
     date = email.utils.formatdate(timeval=None, localtime=False, usegmt=True)
@@ -394,6 +398,16 @@ def generate_vumark_instance(target_id: str) -> Response:
     # ``target_id`` is validated by request validators.
     del target_id
 
+    database = get_database_matching_server_keys(
+        request_headers=dict(request.headers),
+        request_body=request.data,
+        request_method=request.method,
+        request_path=request.path,
+        databases=all_databases,
+    )
+    if not isinstance(database, VuMarkDatabase):
+        raise InvalidTargetTypeError
+
     accept = request.headers.get(key="Accept", default="")
     valid_accept_types: dict[str, bytes] = {
         "image/png": VUMARK_PNG,
@@ -503,6 +517,10 @@ def target_summary(target_id: str) -> Response:
     (target,) = (
         target for target in database.targets if target.target_id == target_id
     )
+    tracking_rating = target.tracking_rating
+    total_recos = target.total_recos
+    current_month_recos = target.current_month_recos
+    previous_month_recos = target.previous_month_recos
     body = {
         "status": target.status,
         "transaction_id": uuid.uuid4().hex,
@@ -511,10 +529,10 @@ def target_summary(target_id: str) -> Response:
         "target_name": target.name,
         "upload_date": target.upload_date.strftime(format="%Y-%m-%d"),
         "active_flag": target.active_flag,
-        "tracking_rating": target.tracking_rating,
-        "total_recos": target.total_recos,
-        "current_month_recos": target.current_month_recos,
-        "previous_month_recos": target.previous_month_recos,
+        "tracking_rating": tracking_rating,
+        "total_recos": total_recos,
+        "current_month_recos": current_month_recos,
+        "previous_month_recos": previous_month_recos,
     }
     date = email.utils.formatdate(timeval=None, localtime=False, usegmt=True)
     headers = {

src/mock_vws/_requests_mock_server/mock_web_services_api.py

@@ -32,17 +32,19 @@
     FailError,
     InvalidAcceptHeaderError,
     InvalidInstanceIdError,
+    InvalidTargetTypeError,
     TargetStatusNotSuccessError,
     TargetStatusProcessingError,
     ValidatorError,
 )
+from mock_vws.database import VuMarkDatabase
 from mock_vws.image_matchers import ImageMatcher
 from mock_vws.target import ImageTarget
 from mock_vws.target_manager import TargetManager
 from mock_vws.target_raters import TargetTrackingRater
 
 if TYPE_CHECKING:
-    from mock_vws.database import CloudDatabase, VuMarkDatabase
+    from mock_vws.database import CloudDatabase
 
 _TARGET_ID_PATTERN = "[A-Za-z0-9]+"
 
@@ -268,7 +270,7 @@ def delete_target(self, request: PreparedRequest) -> _ResponseType:
 
         now = datetime.datetime.now(tz=target.upload_date.tzinfo)
         # See https://github.com/facebook/pyrefly/issues/1897
-        new_target = copy.replace(
+        new_target: ImageTarget = copy.replace(
             target,  # pyrefly: ignore[bad-argument-type]
             delete_date=now,
         )
@@ -324,6 +326,16 @@ def generate_vumark_instance(
                 databases=all_databases,
             )
 
+            database = get_database_matching_server_keys(
+                request_headers=request.headers,
+                request_body=_body_bytes(request=request),
+                request_method=request.method or "",
+                request_path=request.path_url,
+                databases=all_databases,
+            )
+            if not isinstance(database, VuMarkDatabase):
+                raise InvalidTargetTypeError
+
             accept = dict(request.headers).get("Accept", "")
             if accept not in valid_accept_types:
                 raise InvalidAcceptHeaderError
@@ -500,13 +512,16 @@ def get_target(self, request: PreparedRequest) -> _ResponseType:
         target_id = request.path_url.split(sep="/")[-1]
         target = database.get_target(target_id=target_id)
 
+        width = target.width
+        tracking_rating = target.tracking_rating
+        reco_rating = target.reco_rating
         target_record = {
             "target_id": target.target_id,
             "active_flag": target.active_flag,
             "name": target.name,
-            "width": target.width,
-            "tracking_rating": target.tracking_rating,
-            "reco_rating": target.reco_rating,
+            "width": width,
+            "tracking_rating": tracking_rating,
+            "reco_rating": reco_rating,
         }
         date = email.utils.formatdate(
             timeval=None,
@@ -653,17 +668,8 @@ def update_target(self, request: PreparedRequest) -> _ResponseType:
             )
 
         request_json: dict[str, Any] = json.loads(s=request.body or b"")
-        width = request_json.get("width", target.width)
         name = request_json.get("name", target.name)
         active_flag = request_json.get("active_flag", target.active_flag)
-        application_metadata = request_json.get(
-            "application_metadata",
-            target.application_metadata,
-        )
-
-        image_value = target.image_value
-        if "image" in request_json:
-            image_value = base64.b64decode(s=request_json["image"])
 
         if "active_flag" in request_json and active_flag is None:
             fail_exception = FailError(status_code=HTTPStatus.BAD_REQUEST)
@@ -673,6 +679,19 @@ def update_target(self, request: PreparedRequest) -> _ResponseType:
                 fail_exception.response_text,
             )
 
+        gmt = ZoneInfo(key="GMT")
+        last_modified_date = datetime.datetime.now(tz=gmt)
+
+        width = request_json.get("width", target.width)
+        application_metadata = request_json.get(
+            "application_metadata",
+            target.application_metadata,
+        )
+
+        image_value = target.image_value
+        if "image" in request_json:
+            image_value = base64.b64decode(s=request_json["image"])
+
         if (
             "application_metadata" in request_json
             and application_metadata is None
@@ -684,11 +703,8 @@ def update_target(self, request: PreparedRequest) -> _ResponseType:
                 fail_exception.response_text,
             )
 
-        gmt = ZoneInfo(key="GMT")
-        last_modified_date = datetime.datetime.now(tz=gmt)
-
         # See https://github.com/facebook/pyrefly/issues/1897
-        new_target = copy.replace(
+        new_target: ImageTarget = copy.replace(
             target,  # pyrefly: ignore[bad-argument-type]
             name=name,
             width=width,
@@ -755,6 +771,10 @@ def target_summary(self, request: PreparedRequest) -> _ResponseType:
             localtime=False,
             usegmt=True,
         )
+        tracking_rating = target.tracking_rating
+        total_recos = target.total_recos
+        current_month_recos = target.current_month_recos
+        previous_month_recos = target.previous_month_recos
         body = {
             "status": target.status,
             "transaction_id": uuid.uuid4().hex,
@@ -763,10 +783,10 @@ def target_summary(self, request: PreparedRequest) -> _ResponseType:
             "target_name": target.name,
             "upload_date": target.upload_date.strftime(format="%Y-%m-%d"),
             "active_flag": target.active_flag,
-            "tracking_rating": target.tracking_rating,
-            "total_recos": target.total_recos,
-            "current_month_recos": target.current_month_recos,
-            "previous_month_recos": target.previous_month_recos,
+            "tracking_rating": tracking_rating,
+            "total_recos": total_recos,
+            "current_month_recos": current_month_recos,
+            "previous_month_recos": previous_month_recos,
         }
         body_json = json_dump(body=body)
         headers = {

src/mock_vws/_services_validators/exceptions.py

@@ -646,6 +646,46 @@ def __init__(self) -> None:
         }
 
 
+@beartype
+class InvalidTargetTypeError(ValidatorError):
+    """Exception raised when the target type is not valid for the
+    operation.
+    """
+
+    def __init__(self) -> None:
+        """
+        Attributes:
+            status_code: The status code to use in a response if this is
+                raised.
+            response_text: The response text to use in a response if this
+        is
+                raised.
+        """
+        super().__init__()
+        self.status_code = HTTPStatus.UNPROCESSABLE_ENTITY
+        body = {
+            "transaction_id": uuid.uuid4().hex,
+            "result_code": ResultCodes.INVALID_TARGET_TYPE.value,
+        }
+        self.response_text = json_dump(body=body)
+        date = email.utils.formatdate(
+            timeval=None,
+            localtime=False,
+            usegmt=True,
+        )
+        self.headers = {
+            "Connection": "keep-alive",
+            "Content-Type": "application/json",
+            "server": "envoy",
+            "Date": date,
+            "x-envoy-upstream-service-time": "5",
+            "Content-Length": str(object=len(self.response_text)),
+            "strict-transport-security": "max-age=31536000",
+            "x-aws-region": "us-east-2, us-west-2",
+            "x-content-type-options": "nosniff",
+        }
+
+
 @beartype
 class TargetStatusProcessingError(ValidatorError):
     """Exception raised when trying to delete a target which is processing."""