Reduce duplication across assets module

- Extract validate_blake3_hash() into helpers.py, used by upload, schemas, routes
- Extract get_reference_with_owner_check() into queries, used by 4 service functions
- Extract build_prefix_like_conditions() into queries/common.py, used by 3 queries
- Replace 3 inlined tag queries with get_reference_tags() calls
- Consolidate AddTagsDict/RemoveTagsDict TypedDicts into AddTagsResult/RemoveTagsResult
  dataclasses, eliminating manual field copying in tagging.py
- Make iter_row_chunks delegate to iter_chunks
- Inline trivial compute_filename_for_reference wrapper (unused session param)
- Remove mark_assets_missing_outside_prefixes pass-through in bulk_ingest.py
- Clean up unused imports (os, time, dependencies_available)
- Disable assets routes on DB init failure in main.py

Amp-Thread-ID: https://ampcode.com/threads/T-019cb649-dd4e-71ff-9a0e-ae517365207b
Co-authored-by: Amp <amp@ampcode.com>

app/assets/api/routes.py

@@ -17,6 +17,7 @@ from app.assets.api.schemas_in import (
     AssetValidationError,
     UploadError,
 )
+from app.assets.helpers import validate_blake3_hash
 from app.assets.api.upload import (
     delete_temp_file_if_exists,
     parse_multipart_upload,
@@ -89,6 +90,12 @@ def register_assets_routes(
     app.add_routes(ROUTES)
 
 
+def disable_assets_routes() -> None:
+    """Disable asset routes at runtime (e.g. after DB init failure)."""
+    global _ASSETS_ENABLED
+    _ASSETS_ENABLED = False
+
+
 def _build_error_response(
     status: int, code: str, message: str, details: dict | None = None
 ) -> web.Response:
@@ -116,16 +123,9 @@ def _validate_sort_field(requested: str | None) -> str:
 @_require_assets_feature_enabled
 async def head_asset_by_hash(request: web.Request) -> web.Response:
     hash_str = request.match_info.get("hash", "").strip().lower()
-    if not hash_str or ":" not in hash_str:
-        return _build_error_response(
-            400, "INVALID_HASH", "hash must be like 'blake3:<hex>'"
-        )
-    algo, digest = hash_str.split(":", 1)
-    if (
-        algo != "blake3"
-        or not digest
-        or any(c for c in digest if c not in "0123456789abcdef")
-    ):
+    try:
+        hash_str = validate_blake3_hash(hash_str)
+    except ValueError:
         return _build_error_response(
             400, "INVALID_HASH", "hash must be like 'blake3:<hex>'"
         )

app/assets/api/schemas_in.py

@@ -2,6 +2,7 @@ import json
 from dataclasses import dataclass
 from typing import Any, Literal
 
+from app.assets.helpers import validate_blake3_hash
 from pydantic import (
     BaseModel,
     ConfigDict,
@@ -116,15 +117,7 @@ class CreateFromHashBody(BaseModel):
     @field_validator("hash")
     @classmethod
     def _require_blake3(cls, v):
-        s = (v or "").strip().lower()
-        if ":" not in s:
-            raise ValueError("hash must be 'blake3:<hex>'")
-        algo, digest = s.split(":", 1)
-        if algo != "blake3":
-            raise ValueError("only canonical 'blake3:<hex>' is accepted here")
-        if not digest or any(c for c in digest if c not in "0123456789abcdef"):
-            raise ValueError("hash digest must be lowercase hex")
-        return s
+        return validate_blake3_hash(v or "")
 
     @field_validator("tags", mode="before")
     @classmethod
@@ -214,17 +207,10 @@ class UploadAssetSpec(BaseModel):
     def _parse_hash(cls, v):
         if v is None:
             return None
-        s = str(v).strip().lower()
+        s = str(v).strip()
         if not s:
             return None
-        if ":" not in s:
-            raise ValueError("hash must be 'blake3:<hex>'")
-        algo, digest = s.split(":", 1)
-        if algo != "blake3":
-            raise ValueError("only canonical 'blake3:<hex>' is accepted here")
-        if not digest or any(c for c in digest if c not in "0123456789abcdef"):
-            raise ValueError("hash digest must be lowercase hex")
-        return f"{algo}:{digest}"
+        return validate_blake3_hash(s)
 
     @field_validator("tags", mode="before")
     @classmethod

app/assets/api/upload.py

@@ -7,27 +7,18 @@ from aiohttp import web
 
 import folder_paths
 from app.assets.api.schemas_in import ParsedUpload, UploadError
+from app.assets.helpers import validate_blake3_hash
 
 
 def normalize_and_validate_hash(s: str) -> str:
-    """
-    Validate and normalize a hash string.
+    """Validate and normalize a hash string.
 
     Returns canonical 'blake3:<hex>' or raises UploadError.
     """
-    s = s.strip().lower()
-    if not s:
+    try:
+        return validate_blake3_hash(s)
+    except ValueError:
         raise UploadError(400, "INVALID_HASH", "hash must be like 'blake3:<hex>'")
-    if ":" not in s:
-        raise UploadError(400, "INVALID_HASH", "hash must be like 'blake3:<hex>'")
-    algo, digest = s.split(":", 1)
-    if (
-        algo != "blake3"
-        or len(digest) != 64
-        or any(c for c in digest if c not in "0123456789abcdef")
-    ):
-        raise UploadError(400, "INVALID_HASH", "hash must be like 'blake3:<hex>'")
-    return f"{algo}:{digest}"
 
 
 async def parse_multipart_upload(