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 | VuMarkTarget = copy.replace(  # type: ignore[assignment]
         target,  # pyrefly: ignore[bad-argument-type]
         delete_date=now,
     )
@@ -369,32 +375,39 @@ 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)
-    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(
-        target,  # pyrefly: ignore[bad-argument-type]
-        name=name,
-        width=width,
-        active_flag=active_flag,
-        application_metadata=application_metadata,
-        image_value=image_value,
-        last_modified_date=last_modified_date,
-    )
+    if isinstance(target, ImageTarget):
+        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"])
+        # See https://github.com/facebook/pyrefly/issues/1897
+        new_target: ImageTarget | VuMarkTarget = copy.replace(
+            target,  # pyrefly: ignore[bad-argument-type]
+            name=name,
+            width=width,
+            active_flag=active_flag,
+            application_metadata=application_metadata,
+            image_value=image_value,
+            last_modified_date=last_modified_date,
+        )
+    else:
+        # See https://github.com/facebook/pyrefly/issues/1897
+        new_target = copy.replace(
+            target,  # pyrefly: ignore[bad-argument-type]
+            name=name,
+            active_flag=active_flag,
+            last_modified_date=last_modified_date,
+        )
 
     database.targets.remove(target)
     database.targets.add(new_target)

src/mock_vws/_flask_server/vws.py

@@ -32,6 +32,7 @@
     FailError,
     InvalidAcceptHeaderError,
     InvalidInstanceIdError,
+    InvalidTargetTypeError,
     TargetStatusNotSuccessError,
     TargetStatusProcessingError,
     ValidatorError,
@@ -278,13 +279,21 @@ def get_target(target_id: str) -> Response:
         target for target in database.targets if target.target_id == target_id
     )
 
+    if isinstance(target, ImageTarget):
+        width = target.width
+        tracking_rating = target.tracking_rating
+        reco_rating = target.reco_rating
+    else:
+        width = 0.0
+        tracking_rating = -1
+        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 +403,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 +522,16 @@ def target_summary(target_id: str) -> Response:
     (target,) = (
         target for target in database.targets if target.target_id == target_id
     )
+    if isinstance(target, ImageTarget):
+        tracking_rating = target.tracking_rating
+        total_recos = target.total_recos
+        current_month_recos = target.current_month_recos
+        previous_month_recos = target.previous_month_recos
+    else:
+        tracking_rating = -1
+        total_recos = 0
+        current_month_recos = 0
+        previous_month_recos = 0
     body = {
         "status": target.status,
         "transaction_id": uuid.uuid4().hex,
@@ -511,10 +540,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 = {
@@ -561,17 +590,21 @@ def get_duplicates(target_id: str) -> Response:
     )
     other_targets = database.targets - {target}
 
-    similar_targets = [
-        other.target_id
-        for other in other_targets
-        if image_match_checker(
-            first_image_content=target.image_value,
-            second_image_content=other.image_value,
-        )
-        and TargetStatuses.FAILED.value not in {target.status, other.status}
-        and TargetStatuses.PROCESSING.value != other.status
-        and other.active_flag
-    ]
+    similar_targets = []
+    if isinstance(target, ImageTarget):
+        similar_targets = [
+            other.target_id
+            for other in other_targets
+            if isinstance(other, ImageTarget)
+            and image_match_checker(
+                first_image_content=target.image_value,
+                second_image_content=other.image_value,
+            )
+            and TargetStatuses.FAILED.value
+            not in {target.status, other.status}
+            and TargetStatuses.PROCESSING.value != other.status
+            and other.active_flag
+        ]
 
     body = {
         "transaction_id": uuid.uuid4().hex,

src/mock_vws/_query_tools.py

@@ -16,6 +16,7 @@
 from mock_vws._mock_common import json_dump
 from mock_vws.database import CloudDatabase
 from mock_vws.image_matchers import ImageMatcher
+from mock_vws.target import ImageTarget
 
 
 @beartype
@@ -72,7 +73,8 @@ def get_query_match_response_text(
     matching_targets = [
         target
         for target in database.targets
-        if query_match_checker(
+        if isinstance(target, ImageTarget)
+        and query_match_checker(
             first_image_content=target.image_value,
             second_image_content=image_value,
         )