♻️🐛🎨 Renaming and redesign of celery task metadata

packages/celery-library/src/celery_library/backends/redis.py

@@ -6,12 +6,11 @@
 from models_library.progress_bar import ProgressReport
 from pydantic import ValidationError
 from servicelib.celery.models import (
+    ExecutionMetadata,
+    OwnerMetadata,
     Task,
-    TaskFilter,
     TaskID,
     TaskInfoStore,
-    TaskMetadata,
-    Wildcard,
 )
 from servicelib.redis import RedisClientSDK, handle_redis_returns_union_types
 
@@ -35,7 +34,7 @@ def __init__(self, redis_client_sdk: RedisClientSDK) -> None:
     async def create_task(
         self,
         task_id: TaskID,
-        task_metadata: TaskMetadata,
+        task_metadata: ExecutionMetadata,
         expiry: timedelta,
     ) -> None:
         task_key = _build_key(task_id)
@@ -51,7 +50,7 @@ async def create_task(
             expiry,
         )
 
-    async def get_task_metadata(self, task_id: TaskID) -> TaskMetadata | None:
+    async def get_task_metadata(self, task_id: TaskID) -> ExecutionMetadata | None:
         raw_result = await handle_redis_returns_union_types(
             self._redis_client_sdk.redis.hget(
                 _build_key(task_id), _CELERY_TASK_METADATA_KEY
@@ -61,7 +60,7 @@ async def get_task_metadata(self, task_id: TaskID) -> TaskMetadata | None:
             return None
 
         try:
-            return TaskMetadata.model_validate_json(raw_result)
+            return ExecutionMetadata.model_validate_json(raw_result)
         except ValidationError as exc:
             _logger.debug(
                 "Failed to deserialize task metadata for task %s: %s", task_id, f"{exc}"
@@ -85,9 +84,9 @@ async def get_task_progress(self, task_id: TaskID) -> ProgressReport | None:
             )
             return None
 
-    async def list_tasks(self, task_filter: TaskFilter) -> list[Task]:
+    async def list_tasks(self, task_filter: OwnerMetadata) -> list[Task]:
         search_key = _CELERY_TASK_INFO_PREFIX + task_filter.create_task_id(
-            task_uuid=Wildcard()
+            task_uuid="*"
         )
 
         keys: list[str] = []
@@ -112,10 +111,10 @@ async def list_tasks(self, task_filter: TaskFilter) -> list[Task]:
                 continue
 
             with contextlib.suppress(ValidationError):
-                task_metadata = TaskMetadata.model_validate_json(raw_metadata)
+                task_metadata = ExecutionMetadata.model_validate_json(raw_metadata)
                 tasks.append(
                     Task(
-                        uuid=TaskFilter.get_task_uuid(key),
+                        uuid=OwnerMetadata.get_task_uuid(key),
                         metadata=task_metadata,
                     )
                 )

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

@@ -4,7 +4,6 @@
 
 from celery.exceptions import CeleryError  # type: ignore[import-untyped]
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
-    AsyncJobFilter,
     AsyncJobGet,
     AsyncJobId,
     AsyncJobResult,
@@ -17,7 +16,7 @@
     JobNotDoneError,
     JobSchedulerError,
 )
-from servicelib.celery.models import TaskFilter, TaskState
+from servicelib.celery.models import OwnerMetadata, TaskState
 from servicelib.celery.task_manager import TaskManager
 from servicelib.logging_utils import log_catch
 from servicelib.rabbitmq import RPCRouter
@@ -34,14 +33,13 @@
 
 @router.expose(reraise_if_error_type=(JobSchedulerError, JobMissingError))
 async def cancel(
-    task_manager: TaskManager, job_id: AsyncJobId, job_filter: AsyncJobFilter
+    task_manager: TaskManager, job_id: AsyncJobId, owner_metadata: OwnerMetadata
 ):
     assert task_manager  # nosec
-    assert job_filter  # nosec
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
+    assert owner_metadata  # nosec
     try:
         await task_manager.cancel_task(
-            task_filter=task_filter,
+            owner_metadata=owner_metadata,
             task_uuid=job_id,
         )
     except TaskNotFoundError as exc:
@@ -52,15 +50,14 @@ async def cancel(
 
 @router.expose(reraise_if_error_type=(JobSchedulerError, JobMissingError))
 async def status(
-    task_manager: TaskManager, job_id: AsyncJobId, job_filter: AsyncJobFilter
+    task_manager: TaskManager, job_id: AsyncJobId, owner_metadata: OwnerMetadata
 ) -> AsyncJobStatus:
     assert task_manager  # nosec
-    assert job_filter  # nosec
+    assert owner_metadata  # nosec
 
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
     try:
         task_status = await task_manager.get_task_status(
-            task_filter=task_filter,
+            owner_metadata=owner_metadata,
             task_uuid=job_id,
         )
     except TaskNotFoundError as exc:
@@ -85,23 +82,21 @@ async def status(
     )
 )
 async def result(
-    task_manager: TaskManager, job_id: AsyncJobId, job_filter: AsyncJobFilter
+    task_manager: TaskManager, job_id: AsyncJobId, owner_metadata: OwnerMetadata
 ) -> AsyncJobResult:
     assert task_manager  # nosec
     assert job_id  # nosec
-    assert job_filter  # nosec
-
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
+    assert owner_metadata  # nosec
 
     try:
         _status = await task_manager.get_task_status(
-            task_filter=task_filter,
+            owner_metadata=owner_metadata,
             task_uuid=job_id,
         )
         if not _status.is_done:
             raise JobNotDoneError(job_id=job_id)
         _result = await task_manager.get_task_result(
-            task_filter=task_filter,
+            owner_metadata=owner_metadata,
             task_uuid=job_id,
         )
     except TaskNotFoundError as exc:
@@ -134,13 +129,12 @@ async def result(
 
 @router.expose(reraise_if_error_type=(JobSchedulerError,))
 async def list_jobs(
-    task_manager: TaskManager, job_filter: AsyncJobFilter
+    task_manager: TaskManager, owner_metadata: OwnerMetadata
 ) -> list[AsyncJobGet]:
     assert task_manager  # nosec
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
     try:
         tasks = await task_manager.list_tasks(
-            task_filter=task_filter,
+            owner_metadata=owner_metadata,
         )
     except CeleryError as exc:
         raise JobSchedulerError(exc=f"{exc}") from exc

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

@@ -9,11 +9,11 @@
 from models_library.progress_bar import ProgressReport
 from servicelib.celery.models import (
     TASK_DONE_STATES,
+    ExecutionMetadata,
+    OwnerMetadata,
     Task,
-    TaskFilter,
     TaskID,
     TaskInfoStore,
-    TaskMetadata,
     TaskState,
     TaskStatus,
     TaskUUID,
@@ -39,34 +39,34 @@ class CeleryTaskManager:
 
     async def submit_task(
         self,
-        task_metadata: TaskMetadata,
+        execution_metadata: ExecutionMetadata,
         *,
-        task_filter: TaskFilter,
+        owner_metadata: OwnerMetadata,
         **task_params,
     ) -> TaskUUID:
         with log_context(
             _logger,
             logging.DEBUG,
-            msg=f"Submit {task_metadata.name=}: {task_filter=} {task_params=}",
+            msg=f"Submit {execution_metadata.name=}: {owner_metadata=} {task_params=}",
         ):
             task_uuid = uuid4()
-            task_id = task_filter.create_task_id(task_uuid=task_uuid)
+            task_id = owner_metadata.create_task_id(task_uuid=task_uuid)
 
             expiry = (
                 self._celery_settings.CELERY_EPHEMERAL_RESULT_EXPIRES
-                if task_metadata.ephemeral
+                if execution_metadata.ephemeral
                 else self._celery_settings.CELERY_RESULT_EXPIRES
             )
 
             try:
                 await self._task_info_store.create_task(
-                    task_id, task_metadata, expiry=expiry
+                    task_id, execution_metadata, expiry=expiry
                 )
                 self._celery_app.send_task(
-                    task_metadata.name,
+                    execution_metadata.name,
                     task_id=task_id,
                     kwargs={"task_id": task_id} | task_params,
-                    queue=task_metadata.queue.value,
+                    queue=execution_metadata.queue.value,
                 )
             except CeleryError as exc:
                 try:
@@ -78,20 +78,22 @@ async def submit_task(
                         exc_info=True,
                     )
                 raise TaskSubmissionError(
-                    task_name=task_metadata.name,
+                    task_name=execution_metadata.name,
                     task_id=task_id,
                     task_params=task_params,
                 ) from exc
 
             return task_uuid
 
-    async def cancel_task(self, task_filter: TaskFilter, task_uuid: TaskUUID) -> None:
+    async def cancel_task(
+        self, owner_metadata: OwnerMetadata, task_uuid: TaskUUID
+    ) -> None:
         with log_context(
             _logger,
             logging.DEBUG,
-            msg=f"task cancellation: {task_filter=} {task_uuid=}",
+            msg=f"task cancellation: {owner_metadata=} {task_uuid=}",
         ):
-            task_id = task_filter.create_task_id(task_uuid=task_uuid)
+            task_id = owner_metadata.create_task_id(task_uuid=task_uuid)
             if not await self.task_exists(task_id):
                 raise TaskNotFoundError(task_id=task_id)
 
@@ -106,14 +108,14 @@ def _forget_task(self, task_id: TaskID) -> None:
         self._celery_app.AsyncResult(task_id).forget()
 
     async def get_task_result(
-        self, task_filter: TaskFilter, task_uuid: TaskUUID
+        self, owner_metadata: OwnerMetadata, task_uuid: TaskUUID
     ) -> Any:
         with log_context(
             _logger,
             logging.DEBUG,
-            msg=f"Get task result: {task_filter=} {task_uuid=}",
+            msg=f"Get task result: {owner_metadata=} {task_uuid=}",
         ):
-            task_id = task_filter.create_task_id(task_uuid=task_uuid)
+            task_id = owner_metadata.create_task_id(task_uuid=task_uuid)
             if not await self.task_exists(task_id):
                 raise TaskNotFoundError(task_id=task_id)
 
@@ -149,14 +151,14 @@ def _get_task_celery_state(self, task_id: TaskID) -> TaskState:
         return TaskState(self._celery_app.AsyncResult(task_id).state)
 
     async def get_task_status(
-        self, task_filter: TaskFilter, task_uuid: TaskUUID
+        self, owner_metadata: OwnerMetadata, task_uuid: TaskUUID
     ) -> TaskStatus:
         with log_context(
             _logger,
             logging.DEBUG,
-            msg=f"Getting task status: {task_filter=} {task_uuid=}",
+            msg=f"Getting task status: {owner_metadata=} {task_uuid=}",
         ):
-            task_id = task_filter.create_task_id(task_uuid=task_uuid)
+            task_id = owner_metadata.create_task_id(task_uuid=task_uuid)
             if not await self.task_exists(task_id):
                 raise TaskNotFoundError(task_id=task_id)
 
@@ -169,13 +171,13 @@ async def get_task_status(
                 ),
             )
 
-    async def list_tasks(self, task_filter: TaskFilter) -> list[Task]:
+    async def list_tasks(self, owner_metadata: OwnerMetadata) -> list[Task]:
         with log_context(
             _logger,
             logging.DEBUG,
-            msg=f"Listing tasks: {task_filter=}",
+            msg=f"Listing tasks: {owner_metadata=}",
         ):
-            return await self._task_info_store.list_tasks(task_filter)
+            return await self._task_info_store.list_tasks(owner_metadata)
 
     async def set_task_progress(self, task_id: TaskID, report: ProgressReport) -> None:
         await self._task_info_store.set_task_progress(