fix: make SQLite searches Unicode case-insensitive

spoolman/database/database.py

@@ -12,7 +12,7 @@
 from typing import NamedTuple
 
 from scheduler.asyncio.scheduler import Scheduler
-from sqlalchemy import URL
+from sqlalchemy import URL, event
 from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
 
 from spoolman import env
@@ -30,6 +30,16 @@
 BACKUP_NAME = "spoolman.db"
 
 
+def _unicode_lower(value: str | None) -> str | None:
+    """Lowercase SQLite text using Python's Unicode support."""
+    return value.lower() if value is not None else None
+
+
+def _register_sqlite_functions(connection: sqlite3.Connection, _: object) -> None:
+    """Replace SQLite's ASCII-only lower function for each pooled connection."""
+    connection.create_function("lower", 1, _unicode_lower, deterministic=True)
+
+
 class BackupResult(NamedTuple):
     """The outcome of a backup request."""
 
@@ -112,6 +122,8 @@ def connect(self) -> None:
             pool_pre_ping=True,
             **connection_options,
         )
+        if self.connection_url.drivername == "sqlite+aiosqlite":
+            event.listen(self.engine.sync_engine, "connect", _register_sqlite_functions)
         self.session_maker = async_sessionmaker(self.engine, autocommit=False, autoflush=True, expire_on_commit=False)
 
     def backup(self, target_path: str | PathLike[str]) -> None:

tests_integration/tests/search/test_search.py

@@ -16,6 +16,7 @@
 
 VENDOR_NAME = f"SearchVendor{SFX}"
 FILAMENT_NAME = f"SearchFilament{SFX}"
+FILAMENT_NAME_UNICODE = f"Червоний{SFX}"
 FILAMENT_COMMENT = f"fcomment{SFX}"
 SPOOL_LOCATION = f"SearchLoc{SFX}"
 SPOOL_COMMENT = f"scomment{SFX}"
@@ -126,6 +127,18 @@ def test_search_filament_name(data: Fixture):
     assert _has(body["filaments"], "filament", data.filament["id"], "name")
 
 
+def test_search_filament_name_unicode():
+    result = httpx.post(
+        f"{URL}/api/v1/filament",
+        json={"name": FILAMENT_NAME_UNICODE, "density": 1.25, "diameter": 1.75},
+    )
+    result.raise_for_status()
+    filament = result.json()
+    body = _search(FILAMENT_NAME_UNICODE.lower())
+    assert _has(body["filaments"], "filament", filament["id"], "name")
+    httpx.delete(f"{URL}/api/v1/filament/{filament['id']}").raise_for_status()
+
+
 def test_search_vendor_name(data: Fixture):
     body = _search(VENDOR_NAME)
     assert _has(body["vendors"], "vendor", data.vendor["id"], "name")