add test of field_sorting_key

Author: bisgaard-itis

packages/service-library/src/servicelib/celery/models.py

@@ -1,21 +1,40 @@
 import datetime
+from collections.abc import Callable
 from enum import StrEnum
-from typing import Annotated, Protocol, TypeAlias
+from typing import Annotated, Final, Protocol, TypeAlias
 from uuid import UUID
 
 from models_library.progress_bar import ProgressReport
-from pydantic import BaseModel, ConfigDict, StringConstraints
+from pydantic import BaseModel, ConfigDict, Field, StringConstraints
 from pydantic.config import JsonDict
 
 TaskID: TypeAlias = str
 TaskName: TypeAlias = Annotated[
     str, StringConstraints(strip_whitespace=True, min_length=1)
 ]
 TaskUUID: TypeAlias = UUID
+_TASK_ID_KEY_DELIMITATOR: Final[str] = ":"
 
 
 class TaskFilter(BaseModel):
     model_config = ConfigDict(extra="allow")
+    field_sorting_key: Annotated[Callable[[str], int] | None, Field(exclude=True)] = (
+        None
+    )
+
+    def _build_task_id_prefix(self) -> str:
+        filter_dict = self.model_dump()
+        return _TASK_ID_KEY_DELIMITATOR.join(
+            [
+                f"{key}={filter_dict[key]}"
+                for key in sorted(filter_dict, key=self.field_sorting_key)
+            ]
+        )
+
+    def task_id(self, task_uuid: TaskUUID) -> TaskID:
+        return _TASK_ID_KEY_DELIMITATOR.join(
+            [self._build_task_id_prefix(), f"task_uuid={task_uuid}"]
+        )
 
 
 class TaskState(StrEnum):

packages/service-library/tests/test_celery.py

@@ -1,3 +1,6 @@
+from typing import Literal
+
+import pytest
 from faker import Faker
 from servicelib.celery.models import TaskFilter
 
@@ -15,3 +18,23 @@ async def test_task_filter_serialization():
     }
     task_filter = TaskFilter.model_validate(_dict)
     assert task_filter.model_dump() == _dict
+
+
+@pytest.mark.parametrize("key_direction", ["plus", "minus"])
+async def test_task_filter_sorting_key_not_serialized(
+    key_direction: Literal["plus", "minus"],
+):
+
+    keys = ["a", "aa"]
+    key = lambda s: len(s) if key_direction == "plus" else -len(s)
+
+    task_filter = TaskFilter(
+        a=_faker.random_int(),
+        aa=_faker.word(),
+        field_sorting_key=key,
+    )
+    expected_key = ":".join(
+        [f"{k}={getattr(task_filter, k)}" for k in sorted(keys, key=key)]
+    )
+    assert task_filter._build_task_id_prefix() == expected_key
+    assert "field_sorting_key" not in task_filter.model_dump()