further fixes to OwnerMetadata

bisgaard-itis · bisgaard-itis · commit a66d9446bc5a · 2025-09-19T08:52:22.000+02:00
diff --git a/services/api-server/src/simcore_service_api_server/_service_function_jobs_task_client.py b/services/api-server/src/simcore_service_api_server/_service_function_jobs_task_client.py
@@ -47,7 +47,7 @@
     FunctionJobCacheNotFoundError,
 )
 from .models.api_resources import JobLinks
-from .models.domain.celery_models import ApiWorkerTaskFilter
+from .models.domain.celery_models import ApiServerOwnerMetadata
 from .models.schemas.functions import FunctionJobCreationTaskStatus
 from .models.schemas.jobs import JobInputs, JobPricingSpecification
 from .services_http.webserver import AuthSession
@@ -79,7 +79,7 @@ async def _celery_task_status(
 ) -> FunctionJobCreationTaskStatus:
     if job_creation_task_id is None:
         return FunctionJobCreationTaskStatus.NOT_YET_SCHEDULED
-    task_filter = ApiWorkerTaskFilter(
+    task_filter = ApiServerOwnerMetadata(
         user_id=user_id,
         product_name=product_name,
     )
@@ -379,7 +379,7 @@ async def create_function_job_creation_task(
         )
 
         # run function in celery task
-        task_filter = ApiWorkerTaskFilter(
+        task_filter = ApiServerOwnerMetadata(
             user_id=user_identity.user_id, product_name=user_identity.product_name
         )
 
diff --git a/services/api-server/src/simcore_service_api_server/api/routes/tasks.py b/services/api-server/src/simcore_service_api_server/api/routes/tasks.py
@@ -17,7 +17,9 @@
 from models_library.users import UserID
 from servicelib.celery.models import TaskState, TaskUUID
 from servicelib.fastapi.dependencies import get_app
-from simcore_service_api_server.models.domain.celery_models import ApiWorkerTaskFilter
+from simcore_service_api_server.models.domain.celery_models import (
+    ApiServerOwnerMetadata,
+)
 
 from ...models.schemas.base import ApiServerEnvelope
 from ...models.schemas.errors import ErrorGet
@@ -58,7 +60,7 @@ async def list_tasks(
     product_name: Annotated[ProductName, Depends(get_product_name)],
 ):
     task_manager = get_task_manager(app)
-    task_filter = ApiWorkerTaskFilter(
+    task_filter = ApiServerOwnerMetadata(
         user_id=user_id,
         product_name=product_name,
     )
@@ -103,7 +105,7 @@ async def get_task_status(
     product_name: Annotated[ProductName, Depends(get_product_name)],
 ):
     task_manager = get_task_manager(app)
-    task_filter = ApiWorkerTaskFilter(
+    task_filter = ApiServerOwnerMetadata(
         user_id=user_id,
         product_name=product_name,
     )
@@ -141,7 +143,7 @@ async def cancel_task(
     product_name: Annotated[ProductName, Depends(get_product_name)],
 ):
     task_manager = get_task_manager(app)
-    task_filter = ApiWorkerTaskFilter(
+    task_filter = ApiServerOwnerMetadata(
         user_id=user_id,
         product_name=product_name,
     )
@@ -176,7 +178,7 @@ async def get_task_result(
     product_name: Annotated[ProductName, Depends(get_product_name)],
 ):
     task_manager = get_task_manager(app)
-    task_filter = ApiWorkerTaskFilter(
+    task_filter = ApiServerOwnerMetadata(
         user_id=user_id,
         product_name=product_name,
     )
diff --git a/services/api-server/src/simcore_service_api_server/models/domain/celery_models.py b/services/api-server/src/simcore_service_api_server/models/domain/celery_models.py
@@ -33,7 +33,7 @@
 )
 
 
-class ApiWorkerTaskFilter(OwnerMetadata):
+class ApiServerOwnerMetadata(OwnerMetadata):
     user_id: UserID
     product_name: ProductName
     owner: Annotated[
diff --git a/services/api-server/src/simcore_service_api_server/services_rpc/async_jobs.py b/services/api-server/src/simcore_service_api_server/services_rpc/async_jobs.py
@@ -4,7 +4,6 @@
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
     AsyncJobGet,
     AsyncJobId,
-    AsyncJobOwnerMetadata,
     AsyncJobResult,
     AsyncJobStatus,
 )
@@ -15,6 +14,7 @@
     JobSchedulerError,
 )
 from models_library.api_schemas_storage import STORAGE_RPC_NAMESPACE
+from servicelib.celery.models import OwnerMetadata
 from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
 from servicelib.rabbitmq.rpc_interfaces.async_jobs import async_jobs
 
@@ -41,13 +41,13 @@ class AsyncJobClient:
         }
     )
     async def cancel(
-        self, *, job_id: AsyncJobId, job_filter: AsyncJobOwnerMetadata
+        self, *, job_id: AsyncJobId, owner_metadata: OwnerMetadata
     ) -> None:
         return await async_jobs.cancel(
             self._rabbitmq_rpc_client,
             rpc_namespace=STORAGE_RPC_NAMESPACE,
             job_id=job_id,
-            owner_metadata=job_filter,
+            owner_metadata=owner_metadata,
         )
 
     @_exception_mapper(
@@ -56,13 +56,13 @@ async def cancel(
         }
     )
     async def status(
-        self, *, job_id: AsyncJobId, job_filter: AsyncJobOwnerMetadata
+        self, *, job_id: AsyncJobId, owner_metadata: OwnerMetadata
     ) -> AsyncJobStatus:
         return await async_jobs.status(
             self._rabbitmq_rpc_client,
             rpc_namespace=STORAGE_RPC_NAMESPACE,
             job_id=job_id,
-            job_filter=job_filter,
+            owner_metadata=owner_metadata,
         )
 
     @_exception_mapper(
@@ -74,25 +74,23 @@ async def status(
         }
     )
     async def result(
-        self, *, job_id: AsyncJobId, job_filter: AsyncJobOwnerMetadata
+        self, *, job_id: AsyncJobId, owner_metadata: OwnerMetadata
     ) -> AsyncJobResult:
         return await async_jobs.result(
             self._rabbitmq_rpc_client,
             rpc_namespace=STORAGE_RPC_NAMESPACE,
             job_id=job_id,
-            job_filter=job_filter,
+            owner_metadata=owner_metadata,
         )
 
     @_exception_mapper(
         rpc_exception_map={
             JobSchedulerError: TaskSchedulerError,
         }
     )
-    async def list_jobs(
-        self, *, job_filter: AsyncJobOwnerMetadata
-    ) -> list[AsyncJobGet]:
+    async def list_jobs(self, *, owner_metadata: OwnerMetadata) -> list[AsyncJobGet]:
         return await async_jobs.list_jobs(
             self._rabbitmq_rpc_client,
             rpc_namespace=STORAGE_RPC_NAMESPACE,
-            job_filter=job_filter,
+            owner_metadata=owner_metadata,
         )
diff --git a/services/api-server/src/simcore_service_api_server/services_rpc/storage.py b/services/api-server/src/simcore_service_api_server/services_rpc/storage.py
@@ -3,26 +3,21 @@
 
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
     AsyncJobGet,
-    AsyncJobOwnerMetadata,
 )
 from models_library.api_schemas_webserver.storage import PathToExport
 from models_library.products import ProductName
 from models_library.users import UserID
 from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
 from servicelib.rabbitmq.rpc_interfaces.storage import simcore_s3 as storage_rpc
+from simcore_service_api_server.models.domain.celery_models import (
+    ApiServerOwnerMetadata,
+)
 
-from .._meta import APP_NAME
 from ..exceptions.service_errors_utils import service_exception_mapper
 
 _exception_mapper = partial(service_exception_mapper, service_name="Storage")
 
 
-def get_job_filter(user_id: UserID, product_name: ProductName) -> AsyncJobOwnerMetadata:
-    return AsyncJobOwnerMetadata(
-        user_id=user_id, product_name=product_name, owner=APP_NAME
-    )
-
-
 @dataclass(frozen=True, kw_only=True)
 class StorageService:
     _rpc_client: RabbitMQRPCClient
@@ -38,9 +33,8 @@ async def start_data_export(
             self._rpc_client,
             paths_to_export=paths_to_export,
             export_as="download_link",
-            job_filter=get_job_filter(
-                user_id=self._user_id,
-                product_name=self._product_name,
+            owner_metadata=ApiServerOwnerMetadata(
+                user_id=self._user_id, product_name=self._product_name
             ),
         )
         return async_job_get
diff --git a/services/api-server/tests/unit/api_functions/celery/test_functions_celery.py b/services/api-server/tests/unit/api_functions/celery/test_functions_celery.py
@@ -56,7 +56,9 @@
 )
 from simcore_service_api_server.exceptions.backend_errors import BaseBackEndError
 from simcore_service_api_server.models.api_resources import JobLinks
-from simcore_service_api_server.models.domain.celery_models import ApiWorkerTaskFilter
+from simcore_service_api_server.models.domain.celery_models import (
+    ApiServerOwnerMetadata,
+)
 from simcore_service_api_server.models.domain.functions import (
     PreRegisteredFunctionJobData,
 )
@@ -280,7 +282,7 @@ async def test_celery_error_propagation(
     with_api_server_celery_worker: TestWorkController,
 ):
 
-    task_filter = ApiWorkerTaskFilter(
+    task_filter = ApiServerOwnerMetadata(
         user_id=user_identity.user_id,
         product_name=user_identity.product_name,
     )

Original file line number	Diff line number	Diff line change
`@@ -47,7 +47,7 @@`
`47`	`47`	`FunctionJobCacheNotFoundError,`
`48`	`48`	`)`
`49`	`49`	`from .models.api_resources import JobLinks`
`50`		`-from .models.domain.celery_models import ApiWorkerTaskFilter`
	`50`	`+from .models.domain.celery_models import ApiServerOwnerMetadata`
`51`	`51`	`from .models.schemas.functions import FunctionJobCreationTaskStatus`
`52`	`52`	`from .models.schemas.jobs import JobInputs, JobPricingSpecification`
`53`	`53`	`from .services_http.webserver import AuthSession`
`@@ -79,7 +79,7 @@ async def _celery_task_status(`
`79`	`79`	`) -> FunctionJobCreationTaskStatus:`
`80`	`80`	`if job_creation_task_id is None:`
`81`	`81`	`return FunctionJobCreationTaskStatus.NOT_YET_SCHEDULED`
`82`		`- task_filter = ApiWorkerTaskFilter(`
	`82`	`+ task_filter = ApiServerOwnerMetadata(`
`83`	`83`	`user_id=user_id,`
`84`	`84`	`product_name=product_name,`
`85`	`85`	`)`
`@@ -379,7 +379,7 @@ async def create_function_job_creation_task(`
`379`	`379`	`)`
`380`	`380`
`381`	`381`	`# run function in celery task`
`382`		`- task_filter = ApiWorkerTaskFilter(`
	`382`	`+ task_filter = ApiServerOwnerMetadata(`
`383`	`383`	`user_id=user_identity.user_id, product_name=user_identity.product_name`
`384`	`384`	`)`
`385`	`385`
Original file line number	Diff line number	Diff line change
`@@ -33,7 +33,7 @@`
`33`	`33`	`)`
`34`	`34`
`35`	`35`
`36`		`-class ApiWorkerTaskFilter(OwnerMetadata):`
	`36`	`+class ApiServerOwnerMetadata(OwnerMetadata):`
`37`	`37`	`user_id: UserID`
`38`	`38`	`product_name: ProductName`
`39`	`39`	`owner: Annotated[`
Original file line number	Diff line number	Diff line change
`@@ -4,7 +4,6 @@`
`4`	`4`	`from models_library.api_schemas_rpc_async_jobs.async_jobs import (`
`5`	`5`	`AsyncJobGet,`
`6`	`6`	`AsyncJobId,`
`7`		`- AsyncJobOwnerMetadata,`
`8`	`7`	`AsyncJobResult,`
`9`	`8`	`AsyncJobStatus,`
`10`	`9`	`)`
`@@ -15,6 +14,7 @@`
`15`	`14`	`JobSchedulerError,`
`16`	`15`	`)`
`17`	`16`	`from models_library.api_schemas_storage import STORAGE_RPC_NAMESPACE`
	`17`	`+from servicelib.celery.models import OwnerMetadata`
`18`	`18`	`from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient`
`19`	`19`	`from servicelib.rabbitmq.rpc_interfaces.async_jobs import async_jobs`
`20`	`20`
`@@ -41,13 +41,13 @@ class AsyncJobClient:`
`41`	`41`	`}`
`42`	`42`	`)`
`43`	`43`	`async def cancel(`
`44`		`- self, *, job_id: AsyncJobId, job_filter: AsyncJobOwnerMetadata`
	`44`	`+ self, *, job_id: AsyncJobId, owner_metadata: OwnerMetadata`
`45`	`45`	`) -> None:`
`46`	`46`	`return await async_jobs.cancel(`
`47`	`47`	`self._rabbitmq_rpc_client,`
`48`	`48`	`rpc_namespace=STORAGE_RPC_NAMESPACE,`
`49`	`49`	`job_id=job_id,`
`50`		`- owner_metadata=job_filter,`
	`50`	`+ owner_metadata=owner_metadata,`
`51`	`51`	`)`
`52`	`52`
`53`	`53`	`@_exception_mapper(`
`@@ -56,13 +56,13 @@ async def cancel(`
`56`	`56`	`}`
`57`	`57`	`)`
`58`	`58`	`async def status(`
`59`		`- self, *, job_id: AsyncJobId, job_filter: AsyncJobOwnerMetadata`
	`59`	`+ self, *, job_id: AsyncJobId, owner_metadata: OwnerMetadata`
`60`	`60`	`) -> AsyncJobStatus:`
`61`	`61`	`return await async_jobs.status(`
`62`	`62`	`self._rabbitmq_rpc_client,`
`63`	`63`	`rpc_namespace=STORAGE_RPC_NAMESPACE,`
`64`	`64`	`job_id=job_id,`
`65`		`- job_filter=job_filter,`
	`65`	`+ owner_metadata=owner_metadata,`
`66`	`66`	`)`
`67`	`67`
`68`	`68`	`@_exception_mapper(`
`@@ -74,25 +74,23 @@ async def status(`
`74`	`74`	`}`
`75`	`75`	`)`
`76`	`76`	`async def result(`
`77`		`- self, *, job_id: AsyncJobId, job_filter: AsyncJobOwnerMetadata`
	`77`	`+ self, *, job_id: AsyncJobId, owner_metadata: OwnerMetadata`
`78`	`78`	`) -> AsyncJobResult:`
`79`	`79`	`return await async_jobs.result(`
`80`	`80`	`self._rabbitmq_rpc_client,`
`81`	`81`	`rpc_namespace=STORAGE_RPC_NAMESPACE,`
`82`	`82`	`job_id=job_id,`
`83`		`- job_filter=job_filter,`
	`83`	`+ owner_metadata=owner_metadata,`
`84`	`84`	`)`
`85`	`85`
`86`	`86`	`@_exception_mapper(`
`87`	`87`	`rpc_exception_map={`
`88`	`88`	`JobSchedulerError: TaskSchedulerError,`
`89`	`89`	`}`
`90`	`90`	`)`
`91`		`- async def list_jobs(`
`92`		`- self, *, job_filter: AsyncJobOwnerMetadata`
`93`		`- ) -> list[AsyncJobGet]:`
	`91`	`+ async def list_jobs(self, *, owner_metadata: OwnerMetadata) -> list[AsyncJobGet]:`
`94`	`92`	`return await async_jobs.list_jobs(`
`95`	`93`	`self._rabbitmq_rpc_client,`
`96`	`94`	`rpc_namespace=STORAGE_RPC_NAMESPACE,`
`97`		`- job_filter=job_filter,`
	`95`	`+ owner_metadata=owner_metadata,`
`98`	`96`	`)`
Original file line number	Diff line number	Diff line change
`@@ -56,7 +56,9 @@`
`56`	`56`	`)`
`57`	`57`	`from simcore_service_api_server.exceptions.backend_errors import BaseBackEndError`
`58`	`58`	`from simcore_service_api_server.models.api_resources import JobLinks`
`59`		`-from simcore_service_api_server.models.domain.celery_models import ApiWorkerTaskFilter`
	`59`	`+from simcore_service_api_server.models.domain.celery_models import (`
	`60`	`+ ApiServerOwnerMetadata,`
	`61`	`+)`
`60`	`62`	`from simcore_service_api_server.models.domain.functions import (`
`61`	`63`	`PreRegisteredFunctionJobData,`
`62`	`64`	`)`
`@@ -280,7 +282,7 @@ async def test_celery_error_propagation(`
`280`	`282`	`with_api_server_celery_worker: TestWorkController,`
`281`	`283`	`):`
`282`	`284`
`283`		`- task_filter = ApiWorkerTaskFilter(`
	`285`	`+ task_filter = ApiServerOwnerMetadata(`
`284`	`286`	`user_id=user_identity.user_id,`
`285`	`287`	`product_name=user_identity.product_name,`
`286`	`288`	`)`