add task_owner as required field in TaskFilter

Author: bisgaard-itis

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

@@ -44,6 +44,7 @@ class MyTaskFilter(TaskFilter):
     """
 
     model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)
+    task_owner: Annotated[str, StringConstraints(min_length=1, pattern=r"^[a-z_-]+$")]
 
     @model_validator(mode="after")
     def _check_valid_filters(self) -> Self:

packages/service-library/tests/test_celery.py

@@ -1,9 +1,11 @@
+from typing import Annotated
+
 # pylint: disable=redefined-outer-name
 # pylint: disable=protected-access
 import pydantic
 import pytest
 from faker import Faker
-from pydantic import BaseModel
+from pydantic import BaseModel, StringConstraints
 from servicelib.celery.models import TaskFilter, TaskUUID
 
 _faker = Faker()
@@ -80,3 +82,22 @@ class MyModel(BaseModel):
 def test_task_filter_validator_raises_on_forbidden_chars(bad_data):
     with pytest.raises(pydantic.ValidationError):
         TaskFilter.model_validate(bad_data)
+
+
+async def test_task_owner():
+    class MyFilter(TaskFilter):
+        extra_field: str
+
+    with pytest.raises(pydantic.ValidationError):
+        MyFilter(task_owner="", extra_field="value")
+
+    with pytest.raises(pydantic.ValidationError):
+        MyFilter(task_owner="UPPER_CASE", extra_field="value")
+
+    class MyNextFilter(TaskFilter):
+        task_owner: Annotated[
+            str, StringConstraints(strip_whitespace=True, pattern=r"^the_task_owner$")
+        ]
+
+    with pytest.raises(pydantic.ValidationError):
+        MyNextFilter(task_owner="wrong_owner")