further refinements

Author: bisgaard-itis

packages/celery-library/src/celery_library/rpc/_async_jobs.py

@@ -36,9 +36,10 @@ async def cancel(
 ):
     assert task_manager  # nosec
     assert job_filter  # nosec
+    task_filter = TaskFilter.from_async_job_filter(job_filter)
     try:
         await task_manager.cancel_task(
-            task_filter=TaskFilter.model_validate(job_filter.model_dump()),
+            task_filter=task_filter,
             task_uuid=job_id,
         )
     except CeleryError as exc:
@@ -52,9 +53,10 @@ async def status(
     assert task_manager  # nosec
     assert job_filter  # nosec
 
+    task_filter = TaskFilter.from_async_job_filter(job_filter)
     try:
         task_status = await task_manager.get_task_status(
-            task_filter=TaskFilter.model_validate(job_filter.model_dump()),
+            task_filter=task_filter,
             task_uuid=job_id,
         )
     except CeleryError as exc:
@@ -82,7 +84,7 @@ async def result(
     assert job_id  # nosec
     assert job_filter  # nosec
 
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
+    task_filter = TaskFilter.from_async_job_filter(job_filter)
 
     try:
         _status = await task_manager.get_task_status(
@@ -129,9 +131,10 @@ async def list_jobs(
 ) -> list[AsyncJobGet]:
     _ = filter_
     assert task_manager  # nosec
+    task_filter = TaskFilter.from_async_job_filter(job_filter)
     try:
         tasks = await task_manager.list_tasks(
-            task_filter=TaskFilter.model_validate(job_filter.model_dump()),
+            task_filter=task_filter,
         )
     except CeleryError as exc:
         raise JobSchedulerError(exc=f"{exc}") from exc

packages/celery-library/src/celery_library/utils.py

@@ -9,14 +9,14 @@
 _TASK_ID_KEY_DELIMITATOR: Final[str] = ":"
 
 
-def build_task_id_prefix(task_context: TaskFilter) -> str:
-    _dict = task_context.model_dump()
+def build_task_id_prefix(task_filter: TaskFilter) -> str:
+    _dict = task_filter.model_dump()
     return _TASK_ID_KEY_DELIMITATOR.join([f"{_dict[key]}" for key in sorted(_dict)])
 
 
-def build_task_id(task_context: TaskFilter, task_uuid: TaskUUID) -> TaskID:
+def build_task_id(task_filter: TaskFilter, task_uuid: TaskUUID) -> TaskID:
     return _TASK_ID_KEY_DELIMITATOR.join(
-        [build_task_id_prefix(task_context), f"{task_uuid}"]
+        [build_task_id_prefix(task_filter), f"{task_uuid}"]
     )
 
 

packages/celery-library/tests/unit/test_async_jobs.py

@@ -82,7 +82,7 @@ async def rpc_sync_job(
     task_manager: TaskManager, *, job_filter: AsyncJobFilter, **kwargs: Any
 ) -> AsyncJobGet:
     task_name = sync_job.__name__
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
+    task_filter = TaskFilter.from_async_job_filter(job_filter)
     task_uuid = await task_manager.submit_task(
         TaskMetadata(name=task_name), task_filter=task_filter, **kwargs
     )
@@ -95,7 +95,7 @@ async def rpc_async_job(
     task_manager: TaskManager, *, job_filter: AsyncJobFilter, **kwargs: Any
 ) -> AsyncJobGet:
     task_name = async_job.__name__
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
+    task_filter = TaskFilter.from_async_job_filter(job_filter)
     task_uuid = await task_manager.submit_task(
         TaskMetadata(name=task_name), task_filter=task_filter, **kwargs
     )

packages/celery-library/tests/unit/test_tasks.py

@@ -93,13 +93,13 @@ def _(celery_app: Celery) -> None:
 async def test_submitting_task_calling_async_function_results_with_success_state(
     celery_task_manager: CeleryTaskManager,
 ):
-    task_context = TaskFilter(user_id=42)
+    task_filter = TaskFilter(user_id=42)
 
     task_uuid = await celery_task_manager.submit_task(
         TaskMetadata(
             name=fake_file_processor.__name__,
         ),
-        task_context=task_context,
+        task_filter=task_filter,
         files=[f"file{n}" for n in range(5)],
     )
 
@@ -109,27 +109,27 @@ async def test_submitting_task_calling_async_function_results_with_success_state
         stop=stop_after_delay(30),
     ):
         with attempt:
-            status = await celery_task_manager.get_task_status(task_context, task_uuid)
+            status = await celery_task_manager.get_task_status(task_filter, task_uuid)
             assert status.task_state == TaskState.SUCCESS
 
     assert (
-        await celery_task_manager.get_task_status(task_context, task_uuid)
+        await celery_task_manager.get_task_status(task_filter, task_uuid)
     ).task_state == TaskState.SUCCESS
     assert (
-        await celery_task_manager.get_task_result(task_context, task_uuid)
+        await celery_task_manager.get_task_result(task_filter, task_uuid)
     ) == "archive.zip"
 
 
 async def test_submitting_task_with_failure_results_with_error(
     celery_task_manager: CeleryTaskManager,
 ):
-    task_context = TaskFilter(user_id=42)
+    task_filter = TaskFilter(user_id=42)
 
     task_uuid = await celery_task_manager.submit_task(
         TaskMetadata(
             name=failure_task.__name__,
         ),
-        task_context=task_context,
+        task_filter=task_filter,
     )
 
     for attempt in Retrying(
@@ -140,58 +140,56 @@ async def test_submitting_task_with_failure_results_with_error(
 
         with attempt:
             raw_result = await celery_task_manager.get_task_result(
-                task_context, task_uuid
+                task_filter, task_uuid
             )
             assert isinstance(raw_result, TransferrableCeleryError)
 
-    raw_result = await celery_task_manager.get_task_result(task_context, task_uuid)
+    raw_result = await celery_task_manager.get_task_result(task_filter, task_uuid)
     assert f"{raw_result}" == "Something strange happened: BOOM!"
 
 
 async def test_cancelling_a_running_task_aborts_and_deletes(
     celery_task_manager: CeleryTaskManager,
 ):
-    task_context = TaskFilter(user_id=42)
+    task_filter = TaskFilter(user_id=42)
 
     task_uuid = await celery_task_manager.submit_task(
         TaskMetadata(
             name=dreamer_task.__name__,
         ),
-        task_context=task_context,
+        task_filter=task_filter,
     )
 
     await asyncio.sleep(3.0)
 
-    await celery_task_manager.cancel_task(task_context, task_uuid)
+    await celery_task_manager.cancel_task(task_filter, task_uuid)
 
     for attempt in Retrying(
         retry=retry_if_exception_type(AssertionError),
         wait=wait_fixed(1),
         stop=stop_after_delay(30),
     ):
         with attempt:
-            progress = await celery_task_manager.get_task_status(
-                task_context, task_uuid
-            )
+            progress = await celery_task_manager.get_task_status(task_filter, task_uuid)
             assert progress.task_state == TaskState.ABORTED
 
     assert (
-        await celery_task_manager.get_task_status(task_context, task_uuid)
+        await celery_task_manager.get_task_status(task_filter, task_uuid)
     ).task_state == TaskState.ABORTED
 
-    assert task_uuid not in await celery_task_manager.list_tasks(task_context)
+    assert task_uuid not in await celery_task_manager.list_tasks(task_filter)
 
 
 async def test_listing_task_uuids_contains_submitted_task(
     celery_task_manager: CeleryTaskManager,
 ):
-    task_context = TaskFilter(user_id=42)
+    task_filter = TaskFilter(user_id=42)
 
     task_uuid = await celery_task_manager.submit_task(
         TaskMetadata(
             name=dreamer_task.__name__,
         ),
-        task_context=task_context,
+        task_filter=task_filter,
     )
 
     for attempt in Retrying(
@@ -200,8 +198,8 @@ async def test_listing_task_uuids_contains_submitted_task(
         stop=stop_after_delay(10),
     ):
         with attempt:
-            tasks = await celery_task_manager.list_tasks(task_context)
+            tasks = await celery_task_manager.list_tasks(task_filter)
             assert any(task.uuid == task_uuid for task in tasks)
 
-        tasks = await celery_task_manager.list_tasks(task_context)
+        tasks = await celery_task_manager.list_tasks(task_filter)
         assert any(task.uuid == task_uuid for task in tasks)

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

@@ -3,6 +3,7 @@
 from typing import Annotated, Protocol, TypeAlias
 from uuid import UUID
 
+from models_library.api_schemas_rpc_async_jobs.async_jobs import AsyncJobFilter
 from models_library.progress_bar import ProgressReport
 from pydantic import BaseModel, ConfigDict, StringConstraints
 
@@ -16,6 +17,10 @@
 class TaskFilter(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
+    @classmethod
+    def from_async_job_filter(cls, async_job_filter: AsyncJobFilter) -> "TaskFilter":
+        cls.model_validate(async_job_filter.model_dump())
+
 
 class TaskState(StrEnum):
     PENDING = "PENDING"

services/storage/src/simcore_service_storage/api/rpc/_simcore_s3.py

@@ -4,7 +4,7 @@
 )
 from models_library.api_schemas_storage.storage_schemas import FoldersBody
 from models_library.api_schemas_webserver.storage import PathToExport
-from servicelib.celery.models import TaskMetadata, TasksQueue
+from servicelib.celery.models import TaskFilter, TaskMetadata, TasksQueue
 from servicelib.celery.task_manager import TaskManager
 from servicelib.rabbitmq import RPCRouter
 
@@ -20,11 +20,12 @@ async def copy_folders_from_project(
     body: FoldersBody,
 ) -> AsyncJobGet:
     task_name = deep_copy_files_from_project.__name__
+    task_filter = TaskFilter.model_validate(job_filter.model_dump())
     task_uuid = await task_manager.submit_task(
         task_metadata=TaskMetadata(
             name=task_name,
         ),
-        task_filter=job_filter,
+        task_filter=task_filter,
         user_id=job_filter.user_id,
         body=body,
     )