diff --git a/spoolman/database/database.py b/spoolman/database/database.py index 537cbf8c4..e0dc2e1e5 100644 --- a/spoolman/database/database.py +++ b/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: diff --git a/tests_integration/tests/search/test_search.py b/tests_integration/tests/search/test_search.py index ced8aec96..9b6890c57 100644 --- a/tests_integration/tests/search/test_search.py +++ b/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")