get rid of AsyncJobFilter

Author: bisgaard-itis

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

@@ -17,7 +17,6 @@
 from faker import Faker
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
     AsyncJobGet,
-    AsyncJobOwnerMetadata,
 )
 from models_library.api_schemas_rpc_async_jobs.exceptions import (
     JobError,
@@ -79,25 +78,23 @@ def product_name(faker: Faker) -> ProductName:
 
 @router.expose()
 async def rpc_sync_job(
-    task_manager: TaskManager, *, job_filter: AsyncJobOwnerMetadata, **kwargs: Any
+    task_manager: TaskManager, *, owner_metadata: OwnerMetadata, **kwargs: Any
 ) -> AsyncJobGet:
     task_name = sync_job.__name__
-    task_filter = OwnerMetadata.model_validate(job_filter.model_dump())
     task_uuid = await task_manager.submit_task(
-        ExecutionMetadata(name=task_name), owner_metadata=task_filter, **kwargs
+        ExecutionMetadata(name=task_name), owner_metadata=owner_metadata, **kwargs
     )
 
     return AsyncJobGet(job_id=task_uuid, job_name=task_name)
 
 
 @router.expose()
 async def rpc_async_job(
-    task_manager: TaskManager, *, job_filter: AsyncJobOwnerMetadata, **kwargs: Any
+    task_manager: TaskManager, *, owner_metadata: OwnerMetadata, **kwargs: Any
 ) -> AsyncJobGet:
     task_name = async_job.__name__
-    task_filter = OwnerMetadata.model_validate(job_filter.model_dump())
     task_uuid = await task_manager.submit_task(
-        ExecutionMetadata(name=task_name), owner_metadata=task_filter, **kwargs
+        ExecutionMetadata(name=task_name), owner_metadata=owner_metadata, **kwargs
     )
 
     return AsyncJobGet(job_id=task_uuid, job_name=task_name)
@@ -158,15 +155,15 @@ async def _start_task_via_rpc(
     user_id: UserID,
     product_name: ProductName,
     **kwargs: Any,
-) -> tuple[AsyncJobGet, AsyncJobOwnerMetadata]:
-    job_filter = AsyncJobOwnerMetadata(
+) -> tuple[AsyncJobGet, OwnerMetadata]:
+    job_filter = OwnerMetadata(
         user_id=user_id, product_name=product_name, owner="pytest_client"
     )
     async_job_get = await async_jobs.submit(
         rabbitmq_rpc_client=client,
         rpc_namespace=ASYNC_JOBS_RPC_NAMESPACE,
         method_name=rpc_task_name,
-        job_filter=job_filter,
+        owner_metadata=job_filter,
         **kwargs,
     )
     return async_job_get, job_filter
@@ -197,7 +194,7 @@ async def _wait_for_job(
     rpc_client: RabbitMQRPCClient,
     *,
     async_job_get: AsyncJobGet,
-    job_filter: AsyncJobOwnerMetadata,
+    owner_metadata: OwnerMetadata,
     stop_after: timedelta = timedelta(seconds=5),
 ) -> None:
 
@@ -212,7 +209,7 @@ async def _wait_for_job(
                 rpc_client,
                 rpc_namespace=ASYNC_JOBS_RPC_NAMESPACE,
                 job_id=async_job_get.job_id,
-                job_filter=job_filter,
+                owner_metadata=owner_metadata,
             )
             assert (
                 result.done is True
@@ -246,7 +243,7 @@ async def test_async_jobs_workflow(
     exposed_rpc_start: str,
     payload: Any,
 ):
-    async_job_get, job_filter = await _start_task_via_rpc(
+    async_job_get, owner_metadata = await _start_task_via_rpc(
         async_jobs_rabbitmq_rpc_client,
         rpc_task_name=exposed_rpc_start,
         user_id=user_id,
@@ -258,21 +255,21 @@ async def test_async_jobs_workflow(
     jobs = await async_jobs.list_jobs(
         async_jobs_rabbitmq_rpc_client,
         rpc_namespace=ASYNC_JOBS_RPC_NAMESPACE,
-        job_filter=job_filter,
+        owner_metadata=owner_metadata,
     )
     assert len(jobs) > 0
 
     await _wait_for_job(
         async_jobs_rabbitmq_rpc_client,
         async_job_get=async_job_get,
-        job_filter=job_filter,
+        owner_metadata=owner_metadata,
     )
 
     async_job_result = await async_jobs.result(
         async_jobs_rabbitmq_rpc_client,
         rpc_namespace=ASYNC_JOBS_RPC_NAMESPACE,
         job_id=async_job_get.job_id,
-        job_filter=job_filter,
+        owner_metadata=owner_metadata,
     )
     assert async_job_result.result == payload
 
@@ -310,7 +307,7 @@ async def test_async_jobs_cancel(
     jobs = await async_jobs.list_jobs(
         async_jobs_rabbitmq_rpc_client,
         rpc_namespace=ASYNC_JOBS_RPC_NAMESPACE,
-        job_filter=job_filter,
+        owner_metadata=job_filter,
     )
     assert async_job_get.job_id not in [job.job_id for job in jobs]
 
@@ -319,15 +316,15 @@ 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,
         )
 
     with pytest.raises(JobMissingError):
         await async_jobs.result(
             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,
         )
 
 
@@ -357,7 +354,7 @@ async def test_async_jobs_raises(
     exposed_rpc_start: str,
     error: Exception,
 ):
-    async_job_get, job_filter = await _start_task_via_rpc(
+    async_job_get, owner_metadata = await _start_task_via_rpc(
         async_jobs_rabbitmq_rpc_client,
         rpc_task_name=exposed_rpc_start,
         user_id=user_id,
@@ -369,7 +366,7 @@ async def test_async_jobs_raises(
     await _wait_for_job(
         async_jobs_rabbitmq_rpc_client,
         async_job_get=async_job_get,
-        job_filter=job_filter,
+        owner_metadata=owner_metadata,
         stop_after=timedelta(minutes=1),
     )
 
@@ -378,7 +375,7 @@ async def test_async_jobs_raises(
             async_jobs_rabbitmq_rpc_client,
             rpc_namespace=ASYNC_JOBS_RPC_NAMESPACE,
             job_id=async_job_get.job_id,
-            job_filter=job_filter,
+            owner_metadata=owner_metadata,
         )
     assert exc.value.exc_type == type(error).__name__
     assert exc.value.exc_msg == f"{error}"

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

@@ -1,8 +1,6 @@
 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,23 +40,3 @@ class AsyncJobGet(BaseModel):
 class AsyncJobAbort(BaseModel):
     result: bool
     job_id: AsyncJobId
-
-
-class AsyncJobOwnerMetadata(BaseModel):
-    """Data for controlling access to an async job"""
-
-    model_config = ConfigDict(
-        extra="allow",
-        json_schema_extra={
-            "examples": [
-                {
-                    "product_name": "osparc",
-                    "user_id": 123,
-                    "owner": "web_client",
-                }
-            ]
-        },
-    )
-    user_id: UserID
-    product_name: ProductName
-    owner: Annotated[str, StringConstraints(min_length=1, pattern=r"^[a-z_-]+$")]

packages/pytest-simcore/src/pytest_simcore/helpers/async_jobs_server.py

@@ -5,7 +5,6 @@
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
     AsyncJobGet,
     AsyncJobId,
-    AsyncJobOwnerMetadata,
     AsyncJobResult,
     AsyncJobStatus,
 )
@@ -14,6 +13,7 @@
 from models_library.rabbitmq_basic_types import RPCNamespace
 from pydantic import validate_call
 from pytest_mock import MockType
+from servicelib.celery.models import OwnerMetadata
 from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
 
 
@@ -28,7 +28,7 @@ async def cancel(
         *,
         rpc_namespace: RPCNamespace,
         job_id: AsyncJobId,
-        job_filter: AsyncJobOwnerMetadata,
+        owner_metadata: OwnerMetadata,
     ) -> None:
         if self.exception is not None:
             raise self.exception
@@ -41,7 +41,7 @@ async def status(
         *,
         rpc_namespace: RPCNamespace,
         job_id: AsyncJobId,
-        job_filter: AsyncJobOwnerMetadata,
+        owner_metadata: OwnerMetadata,
     ) -> AsyncJobStatus:
         if self.exception is not None:
             raise self.exception
@@ -63,7 +63,7 @@ async def result(
         *,
         rpc_namespace: RPCNamespace,
         job_id: AsyncJobId,
-        job_filter: AsyncJobOwnerMetadata,
+        owner_metadata: OwnerMetadata,
     ) -> AsyncJobResult:
         if self.exception is not None:
             raise self.exception
@@ -75,7 +75,7 @@ async def list_jobs(
         rabbitmq_rpc_client: RabbitMQRPCClient | MockType,
         *,
         rpc_namespace: RPCNamespace,
-        job_filter: AsyncJobOwnerMetadata,
+        job_filter: OwnerMetadata,
         filter_: str = "",
     ) -> list[AsyncJobGet]:
         if self.exception is not None:

packages/pytest-simcore/src/pytest_simcore/helpers/storage_rpc_server.py

@@ -10,11 +10,11 @@
 
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
     AsyncJobGet,
-    AsyncJobOwnerMetadata,
 )
 from models_library.api_schemas_webserver.storage import PathToExport
 from pydantic import TypeAdapter, validate_call
 from pytest_mock import MockType
+from servicelib.celery.models import OwnerMetadata
 from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
 
 
@@ -27,18 +27,18 @@ async def start_export_data(
         *,
         paths_to_export: list[PathToExport],
         export_as: Literal["path", "download_link"],
-        job_filter: AsyncJobOwnerMetadata,
-    ) -> tuple[AsyncJobGet, AsyncJobOwnerMetadata]:
+        owner_metadata: OwnerMetadata,
+    ) -> tuple[AsyncJobGet, OwnerMetadata]:
         assert rabbitmq_rpc_client
-        assert job_filter
+        assert owner_metadata
         assert paths_to_export
         assert export_as
 
         async_job_get = TypeAdapter(AsyncJobGet).validate_python(
             AsyncJobGet.model_json_schema()["examples"][0],
         )
-        async_job_filter = TypeAdapter(AsyncJobOwnerMetadata).validate_python(
-            AsyncJobOwnerMetadata.model_json_schema()["examples"][0],
+        async_job_filter = TypeAdapter(OwnerMetadata).validate_python(
+            OwnerMetadata.model_json_schema()["examples"][0],
         )
 
         return async_job_get, async_job_filter

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/storage/paths.py

@@ -6,6 +6,7 @@
 from models_library.api_schemas_storage import STORAGE_RPC_NAMESPACE
 from models_library.projects_nodes_io import LocationID
 from models_library.rabbitmq_basic_types import RPCMethodName
+from models_library.users import UserID
 from pydantic import TypeAdapter
 
 from ....celery.models import OwnerMetadata
@@ -19,6 +20,7 @@ async def compute_path_size(
     location_id: LocationID,
     path: Path,
     owner_metadata: OwnerMetadata,
+    user_id: UserID
 ) -> tuple[AsyncJobGet, OwnerMetadata]:
     async_job_rpc_get = await submit(
         rabbitmq_rpc_client=client,
@@ -27,6 +29,7 @@ async def compute_path_size(
         owner_metadata=owner_metadata,
         location_id=location_id,
         path=path,
+        user_id=user_id,
     )
     return async_job_rpc_get, owner_metadata
 
@@ -37,6 +40,7 @@ async def delete_paths(
     location_id: LocationID,
     paths: set[Path],
     owner_metadata: OwnerMetadata,
+    user_id: UserID
 ) -> tuple[AsyncJobGet, OwnerMetadata]:
     async_job_rpc_get = await submit(
         rabbitmq_rpc_client=client,
@@ -45,5 +49,6 @@ async def delete_paths(
         owner_metadata=owner_metadata,
         location_id=location_id,
         paths=paths,
+        user_id=user_id,
     )
     return async_job_rpc_get, owner_metadata

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/storage/simcore_s3.py

@@ -7,6 +7,7 @@
 from models_library.api_schemas_storage.storage_schemas import FoldersBody
 from models_library.api_schemas_webserver.storage import PathToExport
 from models_library.rabbitmq_basic_types import RPCMethodName
+from models_library.users import UserID
 from pydantic import TypeAdapter
 from servicelib.celery.models import OwnerMetadata
 
@@ -15,7 +16,11 @@
 
 
 async def copy_folders_from_project(
-    client: RabbitMQRPCClient, *, body: FoldersBody, owner_metadata: OwnerMetadata
+    client: RabbitMQRPCClient,
+    *,
+    body: FoldersBody,
+    owner_metadata: OwnerMetadata,
+    user_id: UserID
 ) -> tuple[AsyncJobGet, OwnerMetadata]:
     async_job_rpc_get = await submit(
         rabbitmq_rpc_client=client,
@@ -25,6 +30,7 @@ async def copy_folders_from_project(
         ),
         owner_metadata=owner_metadata,
         body=body,
+        user_id=user_id,
     )
     return async_job_rpc_get, owner_metadata
 
@@ -34,7 +40,8 @@ async def start_export_data(
     *,
     paths_to_export: list[PathToExport],
     export_as: Literal["path", "download_link"],
-    owner_metadata: OwnerMetadata
+    owner_metadata: OwnerMetadata,
+    user_id: UserID
 ) -> tuple[AsyncJobGet, OwnerMetadata]:
     async_job_rpc_get = await submit(
         rabbitmq_rpc_client,
@@ -43,5 +50,6 @@ async def start_export_data(
         owner_metadata=owner_metadata,
         paths_to_export=paths_to_export,
         export_as=export_as,
+        user_id=user_id,
     )
     return async_job_rpc_get, owner_metadata