improve naming

Author: bisgaard-itis

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

@@ -7,10 +7,10 @@
 from pydantic import ValidationError
 from servicelib.celery.models import (
     Task,
-    TaskFilter,
+    TaskExecutionMetadata,
     TaskID,
     TaskInfoStore,
-    TaskMetadata,
+    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: TaskMetadata,
+        task_metadata: TaskExecutionMetadata,
         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) -> TaskMetadata | None:
+    async def get_task_metadata(self, task_id: TaskID) -> TaskExecutionMetadata | 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) -> TaskMetadata | None:
             return None
 
         try:
-            return TaskMetadata.model_validate_json(raw_result)
+            return TaskExecutionMetadata.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: TaskFilter) -> list[Task]:
+    async def list_tasks(self, task_filter: TaskOwnerMetadata) -> 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: TaskFilter) -> list[Task]:
                 continue
 
             with contextlib.suppress(ValidationError):
-                task_metadata = TaskMetadata.model_validate_json(raw_metadata)
+                task_metadata = TaskExecutionMetadata.model_validate_json(raw_metadata)
                 tasks.append(
                     Task(
-                        uuid=TaskFilter.get_task_uuid(key),
+                        uuid=TaskOwnerMetadata.get_task_uuid(key),
                         metadata=task_metadata,
                     )
                 )

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

@@ -4,9 +4,9 @@
 
 from celery.exceptions import CeleryError  # type: ignore[import-untyped]
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
-    AsyncJobFilter,
     AsyncJobGet,
     AsyncJobId,
+    AsyncJobOwnerMetadata,
     AsyncJobResult,
     AsyncJobStatus,
 )
@@ -17,7 +17,7 @@
     JobNotDoneError,
     JobSchedulerError,
 )
-from servicelib.celery.models import TaskFilter, TaskState
+from servicelib.celery.models import TaskOwnerMetadata, TaskState
 from servicelib.celery.task_manager import TaskManager
 from servicelib.logging_utils import log_catch
 from servicelib.rabbitmq import RPCRouter
@@ -34,11 +34,11 @@
 
 @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, job_filter: AsyncJobOwnerMetadata
 ):
     assert task_manager  # nosec
     assert job_filter  # nosec
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
+    task_filter = TaskOwnerMetadata.model_validate(job_filter.model_dump())
     try:
         await task_manager.cancel_task(
             task_filter=task_filter,
@@ -52,12 +52,12 @@ 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, job_filter: AsyncJobOwnerMetadata
 ) -> AsyncJobStatus:
     assert task_manager  # nosec
     assert job_filter  # nosec
 
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
+    task_filter = TaskOwnerMetadata.model_validate(job_filter.model_dump())
     try:
         task_status = await task_manager.get_task_status(
             task_filter=task_filter,
@@ -85,13 +85,13 @@ async def status(
     )
 )
 async def result(
-    task_manager: TaskManager, job_id: AsyncJobId, job_filter: AsyncJobFilter
+    task_manager: TaskManager, job_id: AsyncJobId, job_filter: AsyncJobOwnerMetadata
 ) -> AsyncJobResult:
     assert task_manager  # nosec
     assert job_id  # nosec
     assert job_filter  # nosec
 
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
+    task_filter = TaskOwnerMetadata.model_validate(job_filter.model_dump())
 
     try:
         _status = await task_manager.get_task_status(
@@ -134,10 +134,10 @@ async def result(
 
 @router.expose(reraise_if_error_type=(JobSchedulerError,))
 async def list_jobs(
-    task_manager: TaskManager, job_filter: AsyncJobFilter
+    task_manager: TaskManager, job_filter: AsyncJobOwnerMetadata
 ) -> list[AsyncJobGet]:
     assert task_manager  # nosec
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
+    task_filter = TaskOwnerMetadata.model_validate(job_filter.model_dump())
     try:
         tasks = await task_manager.list_tasks(
             task_filter=task_filter,

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

@@ -10,10 +10,10 @@
 from servicelib.celery.models import (
     TASK_DONE_STATES,
     Task,
-    TaskFilter,
+    TaskExecutionMetadata,
     TaskID,
     TaskInfoStore,
-    TaskMetadata,
+    TaskOwnerMetadata,
     TaskState,
     TaskStatus,
     TaskUUID,
@@ -39,9 +39,9 @@ class CeleryTaskManager:
 
     async def submit_task(
         self,
-        task_metadata: TaskMetadata,
+        task_metadata: TaskExecutionMetadata,
         *,
-        task_filter: TaskFilter,
+        task_filter: TaskOwnerMetadata,
         **task_params,
     ) -> TaskUUID:
         with log_context(
@@ -85,7 +85,9 @@ async def submit_task(
 
             return task_uuid
 
-    async def cancel_task(self, task_filter: TaskFilter, task_uuid: TaskUUID) -> None:
+    async def cancel_task(
+        self, task_filter: TaskOwnerMetadata, task_uuid: TaskUUID
+    ) -> None:
         with log_context(
             _logger,
             logging.DEBUG,
@@ -106,7 +108,7 @@ 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, task_filter: TaskOwnerMetadata, task_uuid: TaskUUID
     ) -> Any:
         with log_context(
             _logger,
@@ -149,7 +151,7 @@ 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, task_filter: TaskOwnerMetadata, task_uuid: TaskUUID
     ) -> TaskStatus:
         with log_context(
             _logger,
@@ -169,7 +171,7 @@ async def get_task_status(
                 ),
             )
 
-    async def list_tasks(self, task_filter: TaskFilter) -> list[Task]:
+    async def list_tasks(self, task_filter: TaskOwnerMetadata) -> list[Task]:
         with log_context(
             _logger,
             logging.DEBUG,

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

@@ -16,8 +16,8 @@
 from common_library.errors_classes import OsparcErrorMixin
 from faker import Faker
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
-    AsyncJobFilter,
     AsyncJobGet,
+    AsyncJobOwnerMetadata,
 )
 from models_library.api_schemas_rpc_async_jobs.exceptions import (
     JobError,
@@ -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 TaskFilter, TaskID, TaskMetadata
+from servicelib.celery.models import TaskExecutionMetadata, TaskID, TaskOwnerMetadata
 from servicelib.celery.task_manager import TaskManager
 from servicelib.rabbitmq import RabbitMQRPCClient, RPCRouter
 from servicelib.rabbitmq.rpc_interfaces.async_jobs import async_jobs
@@ -79,25 +79,25 @@ def product_name(faker: Faker) -> ProductName:
 
 @router.expose()
 async def rpc_sync_job(
-    task_manager: TaskManager, *, job_filter: AsyncJobFilter, **kwargs: Any
+    task_manager: TaskManager, *, job_filter: AsyncJobOwnerMetadata, **kwargs: Any
 ) -> AsyncJobGet:
     task_name = sync_job.__name__
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
+    task_filter = TaskOwnerMetadata.model_validate(job_filter.model_dump())
     task_uuid = await task_manager.submit_task(
-        TaskMetadata(name=task_name), task_filter=task_filter, **kwargs
+        TaskExecutionMetadata(name=task_name), task_filter=task_filter, **kwargs
     )
 
     return AsyncJobGet(job_id=task_uuid, job_name=task_name)
 
 
 @router.expose()
 async def rpc_async_job(
-    task_manager: TaskManager, *, job_filter: AsyncJobFilter, **kwargs: Any
+    task_manager: TaskManager, *, job_filter: AsyncJobOwnerMetadata, **kwargs: Any
 ) -> AsyncJobGet:
     task_name = async_job.__name__
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
+    task_filter = TaskOwnerMetadata.model_validate(job_filter.model_dump())
     task_uuid = await task_manager.submit_task(
-        TaskMetadata(name=task_name), task_filter=task_filter, **kwargs
+        TaskExecutionMetadata(name=task_name), task_filter=task_filter, **kwargs
     )
 
     return AsyncJobGet(job_id=task_uuid, job_name=task_name)
@@ -158,8 +158,8 @@ async def _start_task_via_rpc(
     user_id: UserID,
     product_name: ProductName,
     **kwargs: Any,
-) -> tuple[AsyncJobGet, AsyncJobFilter]:
-    job_filter = AsyncJobFilter(
+) -> tuple[AsyncJobGet, AsyncJobOwnerMetadata]:
+    job_filter = AsyncJobOwnerMetadata(
         user_id=user_id, product_name=product_name, task_owner="pytest_client"
     )
     async_job_get = await async_jobs.submit(
@@ -197,7 +197,7 @@ async def _wait_for_job(
     rpc_client: RabbitMQRPCClient,
     *,
     async_job_get: AsyncJobGet,
-    job_filter: AsyncJobFilter,
+    job_filter: AsyncJobOwnerMetadata,
     stop_after: timedelta = timedelta(seconds=5),
 ) -> None:
 

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

@@ -21,9 +21,9 @@
 from faker import Faker
 from models_library.progress_bar import ProgressReport
 from servicelib.celery.models import (
-    TaskFilter,
+    TaskExecutionMetadata,
     TaskID,
-    TaskMetadata,
+    TaskOwnerMetadata,
     TaskState,
     TaskUUID,
     Wildcard,
@@ -39,7 +39,7 @@
 pytest_simcore_ops_services_selection = []
 
 
-class MyTaskFilter(TaskFilter):
+class MyTaskFilter(TaskOwnerMetadata):
     user_id: int
 
 
@@ -106,7 +106,7 @@ async def test_submitting_task_calling_async_function_results_with_success_state
     task_filter = MyTaskFilter(user_id=42)
 
     task_uuid = await celery_task_manager.submit_task(
-        TaskMetadata(
+        TaskExecutionMetadata(
             name=fake_file_processor.__name__,
         ),
         task_filter=task_filter,
@@ -137,7 +137,7 @@ async def test_submitting_task_with_failure_results_with_error(
     task_filter = MyTaskFilter(user_id=42)
 
     task_uuid = await celery_task_manager.submit_task(
-        TaskMetadata(
+        TaskExecutionMetadata(
             name=failure_task.__name__,
         ),
         task_filter=task_filter,
@@ -166,7 +166,7 @@ async def test_cancelling_a_running_task_aborts_and_deletes(
     task_filter = MyTaskFilter(user_id=42)
 
     task_uuid = await celery_task_manager.submit_task(
-        TaskMetadata(
+        TaskExecutionMetadata(
             name=dreamer_task.__name__,
         ),
         task_filter=task_filter,
@@ -189,7 +189,7 @@ async def test_listing_task_uuids_contains_submitted_task(
     task_filter = MyTaskFilter(user_id=42)
 
     task_uuid = await celery_task_manager.submit_task(
-        TaskMetadata(
+        TaskExecutionMetadata(
             name=dreamer_task.__name__,
         ),
         task_filter=task_filter,
@@ -212,7 +212,7 @@ async def test_filtering_listing_tasks(
     celery_task_manager: CeleryTaskManager,
     with_celery_worker: WorkController,
 ):
-    class MyFilter(TaskFilter):
+    class MyFilter(TaskOwnerMetadata):
         user_id: int
         product_name: str | Wildcard
         client_app: str | Wildcard
@@ -229,7 +229,7 @@ class MyFilter(TaskFilter):
                 client_app=_faker.word(),
             )
             task_uuid = await celery_task_manager.submit_task(
-                TaskMetadata(
+                TaskExecutionMetadata(
                     name=dreamer_task.__name__,
                 ),
                 task_filter=task_filter,
@@ -244,7 +244,7 @@ class MyFilter(TaskFilter):
                 client_app=_faker.word(),
             )
             task_uuid = await celery_task_manager.submit_task(
-                TaskMetadata(
+                TaskExecutionMetadata(
                     name=dreamer_task.__name__,
                 ),
                 task_filter=task_filter,

packages/models-library/src/models_library/api_schemas_rpc_async_jobs/async_jobs.py

@@ -1,6 +1,8 @@
 from typing import Annotated, Any, TypeAlias
 from uuid import UUID
 
+from models_library.products import ProductName
+from models_library.users import UserID
 from pydantic import BaseModel, ConfigDict, StringConstraints
 
 from ..progress_bar import ProgressReport
@@ -42,7 +44,7 @@ class AsyncJobAbort(BaseModel):
     job_id: AsyncJobId
 
 
-class AsyncJobFilter(BaseModel):
+class AsyncJobOwnerMetadata(BaseModel):
     """Data for controlling access to an async job"""
 
     model_config = ConfigDict(
@@ -57,5 +59,6 @@ class AsyncJobFilter(BaseModel):
             ]
         },
     )
-
+    user_id: UserID
+    product_name: ProductName
     task_owner: Annotated[str, StringConstraints(min_length=1, pattern=r"^[a-z_-]+$")]