cleanup in api-server

Author: bisgaard-itis

services/api-server/src/simcore_service_api_server/api/routes/function_jobs_routes.py

@@ -29,6 +29,7 @@
 from ..._service_function_jobs import FunctionJobService
 from ..._service_functions import FunctionService
 from ..._service_jobs import JobService
+from ...clients.celery_task_manager import get_task_filter
 from ...exceptions.function_errors import FunctionJobProjectMissingError
 from ...models.domain.functions import PageRegisteredFunctionJobWithorWithoutStatus
 from ...models.pagination import PaginationParams
@@ -55,7 +56,6 @@
     FMSG_CHANGELOG_NEW_IN_VERSION,
     create_route_description,
 )
-from .tasks import _get_task_filter
 
 _logger = getLogger(__name__)
 
@@ -294,7 +294,7 @@ async def function_job_status(
         ):
             if task_id := function_job.job_creation_task_id:
                 task_manager = get_task_manager(app)
-                task_filter = _get_task_filter(user_id, product_name)
+                task_filter = get_task_filter(user_id, product_name)
                 task_status = await task_manager.get_task_status(
                     task_uuid=TaskUUID(task_id), task_filter=task_filter
                 )

services/api-server/src/simcore_service_api_server/clients/celery_task_manager.py

@@ -11,14 +11,10 @@
 from ..celery_worker.worker_tasks.tasks import pydantic_types_to_register
 
 
-def get_job_filter(user_id: UserID, product_name: ProductName) -> AsyncJobFilter:
-    return AsyncJobFilter(
+def get_task_filter(user_id: UserID, product_name: ProductName) -> TaskFilter:
+    job_filter = AsyncJobFilter(
         user_id=user_id, product_name=product_name, client_name=APP_NAME
     )
-
-
-def get_task_filter(user_id: UserID, product_name: ProductName) -> TaskFilter:
-    job_filter = get_job_filter(user_id=user_id, product_name=product_name)
     return TaskFilter.model_validate(job_filter.model_dump())
 
 

services/api-server/src/simcore_service_api_server/services_rpc/storage.py

@@ -2,6 +2,7 @@
 from functools import partial
 
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
+    AsyncJobFilter,
     AsyncJobGet,
 )
 from models_library.api_schemas_webserver.storage import PathToExport
@@ -10,12 +11,18 @@
 from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
 from servicelib.rabbitmq.rpc_interfaces.storage import simcore_s3 as storage_rpc
 
-from ..clients.celery_task_manager import get_job_filter
+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) -> AsyncJobFilter:
+    return AsyncJobFilter(
+        user_id=user_id, product_name=product_name, client_name=APP_NAME
+    )
+
+
 @dataclass(frozen=True, kw_only=True)
 class StorageService:
     _rpc_client: RabbitMQRPCClient

services/api-server/tests/unit/api_functions/celery/test_functions_celery.py

@@ -54,7 +54,6 @@
 from simcore_service_api_server.celery_worker.worker_tasks.functions_tasks import (
     run_function as run_function_task,
 )
-from simcore_service_api_server.clients.celery_task_manager import get_job_filter
 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.functions import (
@@ -64,6 +63,7 @@
     JobPricingSpecification,
     NodeID,
 )
+from simcore_service_api_server.services_rpc.storage import get_job_filter
 from tenacity import (
     AsyncRetrying,
     retry_if_exception_type,