TaskFilter -> OwnerMetadata

Author: bisgaard-itis

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

@@ -6,11 +6,11 @@
 from models_library.progress_bar import ProgressReport
 from pydantic import ValidationError
 from servicelib.celery.models import (
+    ExecutionMetadata,
+    OwnerMetadata,
     Task,
-    TaskExecutionMetadata,
     TaskID,
     TaskInfoStore,
-    TaskOwnerMetadata,
     Wildcard,
 )
 from servicelib.redis import RedisClientSDK, handle_redis_returns_union_types
@@ -35,7 +35,7 @@ def __init__(self, redis_client_sdk: RedisClientSDK) -> None:
     async def create_task(
         self,
         task_id: TaskID,
-        task_metadata: TaskExecutionMetadata,
+        task_metadata: ExecutionMetadata,
         expiry: timedelta,
     ) -> None:
         task_key = _build_key(task_id)
@@ -51,7 +51,7 @@ async def create_task(
             expiry,
         )
 
-    async def get_task_metadata(self, task_id: TaskID) -> TaskExecutionMetadata | 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 +61,7 @@ async def get_task_metadata(self, task_id: TaskID) -> TaskExecutionMetadata | No
             return None
 
         try:
-            return TaskExecutionMetadata.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,7 +85,7 @@ async def get_task_progress(self, task_id: TaskID) -> ProgressReport | None:
             )
             return None
 
-    async def list_tasks(self, task_filter: TaskOwnerMetadata) -> 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()
         )
@@ -112,10 +112,10 @@ async def list_tasks(self, task_filter: TaskOwnerMetadata) -> list[Task]:
                 continue
 
             with contextlib.suppress(ValidationError):
-                task_metadata = TaskExecutionMetadata.model_validate_json(raw_metadata)
+                task_metadata = ExecutionMetadata.model_validate_json(raw_metadata)
                 tasks.append(
                     Task(
-                        uuid=TaskOwnerMetadata.get_task_uuid(key),
+                        uuid=OwnerMetadata.get_task_uuid(key),
                         metadata=task_metadata,
                     )
                 )

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

@@ -6,7 +6,6 @@
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
     AsyncJobGet,
     AsyncJobId,
-    AsyncJobOwnerMetadata,
     AsyncJobResult,
     AsyncJobStatus,
 )
@@ -17,7 +16,7 @@
     JobNotDoneError,
     JobSchedulerError,
 )
-from servicelib.celery.models import TaskOwnerMetadata, 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: AsyncJobOwnerMetadata
+    task_manager: TaskManager, job_id: AsyncJobId, owner_metadata: OwnerMetadata
 ):
     assert task_manager  # nosec
-    assert job_filter  # nosec
-    task_filter = TaskOwnerMetadata.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: AsyncJobOwnerMetadata
+    task_manager: TaskManager, job_id: AsyncJobId, owner_metadata: OwnerMetadata
 ) -> AsyncJobStatus:
     assert task_manager  # nosec
-    assert job_filter  # nosec
+    assert owner_metadata  # nosec
 
-    task_filter = TaskOwnerMetadata.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: AsyncJobOwnerMetadata
+    task_manager: TaskManager, job_id: AsyncJobId, owner_metadata: OwnerMetadata
 ) -> AsyncJobResult:
     assert task_manager  # nosec
     assert job_id  # nosec
-    assert job_filter  # nosec
-
-    task_filter = TaskOwnerMetadata.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: AsyncJobOwnerMetadata
+    task_manager: TaskManager, owner_metadata: OwnerMetadata
 ) -> list[AsyncJobGet]:
     assert task_manager  # nosec
-    task_filter = TaskOwnerMetadata.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,
-    TaskExecutionMetadata,
     TaskID,
     TaskInfoStore,
-    TaskOwnerMetadata,
     TaskState,
     TaskStatus,
     TaskUUID,
@@ -39,34 +39,34 @@ class CeleryTaskManager:
 
     async def submit_task(
         self,
-        task_metadata: TaskExecutionMetadata,
+        execution_metadata: ExecutionMetadata,
         *,
-        task_filter: TaskOwnerMetadata,
+        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,22 +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: TaskOwnerMetadata, task_uuid: TaskUUID
+        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)
 
@@ -108,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: TaskOwnerMetadata, 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)
 
@@ -151,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: TaskOwnerMetadata, 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)
 
@@ -171,13 +171,13 @@ async def get_task_status(
                 ),
             )
 
-    async def list_tasks(self, task_filter: TaskOwnerMetadata) -> 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(

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

@@ -27,7 +27,7 @@
 from models_library.rabbitmq_basic_types import RPCNamespace
 from models_library.users import UserID
 from pydantic import TypeAdapter
-from servicelib.celery.models import TaskExecutionMetadata, TaskID, TaskOwnerMetadata
+from servicelib.celery.models import ExecutionMetadata, OwnerMetadata, TaskID
 from servicelib.celery.task_manager import TaskManager
 from servicelib.rabbitmq import RabbitMQRPCClient, RPCRouter
 from servicelib.rabbitmq.rpc_interfaces.async_jobs import async_jobs
@@ -82,9 +82,9 @@ async def rpc_sync_job(
     task_manager: TaskManager, *, job_filter: AsyncJobOwnerMetadata, **kwargs: Any
 ) -> AsyncJobGet:
     task_name = sync_job.__name__
-    task_filter = TaskOwnerMetadata.model_validate(job_filter.model_dump())
+    task_filter = OwnerMetadata.model_validate(job_filter.model_dump())
     task_uuid = await task_manager.submit_task(
-        TaskExecutionMetadata(name=task_name), task_filter=task_filter, **kwargs
+        ExecutionMetadata(name=task_name), task_filter=task_filter, **kwargs
     )
 
     return AsyncJobGet(job_id=task_uuid, job_name=task_name)
@@ -95,9 +95,9 @@ async def rpc_async_job(
     task_manager: TaskManager, *, job_filter: AsyncJobOwnerMetadata, **kwargs: Any
 ) -> AsyncJobGet:
     task_name = async_job.__name__
-    task_filter = TaskOwnerMetadata.model_validate(job_filter.model_dump())
+    task_filter = OwnerMetadata.model_validate(job_filter.model_dump())
     task_uuid = await task_manager.submit_task(
-        TaskExecutionMetadata(name=task_name), task_filter=task_filter, **kwargs
+        ExecutionMetadata(name=task_name), task_filter=task_filter, **kwargs
     )
 
     return AsyncJobGet(job_id=task_uuid, job_name=task_name)
@@ -160,7 +160,7 @@ async def _start_task_via_rpc(
     **kwargs: Any,
 ) -> tuple[AsyncJobGet, AsyncJobOwnerMetadata]:
     job_filter = AsyncJobOwnerMetadata(
-        user_id=user_id, product_name=product_name, task_owner="pytest_client"
+        user_id=user_id, product_name=product_name, owner="pytest_client"
     )
     async_job_get = await async_jobs.submit(
         rabbitmq_rpc_client=client,
@@ -304,7 +304,7 @@ async def test_async_jobs_cancel(
         async_jobs_rabbitmq_rpc_client,
         rpc_namespace=ASYNC_JOBS_RPC_NAMESPACE,
         job_id=async_job_get.job_id,
-        job_filter=job_filter,
+        owner_metadata=job_filter,
     )
 
     jobs = await async_jobs.list_jobs(