use app-name for ASYNC_JOB_CLIENT_NAME in api-server

Author: bisgaard-itis

services/api-server/src/simcore_service_api_server/api/dependencies/celery.py

@@ -1,10 +1,6 @@
-from typing import Final
-
 from celery_library.task_manager import CeleryTaskManager
 from fastapi import FastAPI
 
-ASYNC_JOB_CLIENT_NAME: Final[str] = "api-server"
-
 
 def get_task_manager(app: FastAPI) -> CeleryTaskManager:
     assert hasattr(app.state, "task_manager")  # nosec

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

@@ -18,13 +18,12 @@
     RegisteredFunctionJob,
     RegisteredFunctionJobCollection,
 )
-from models_library.api_schemas_rpc_async_jobs.async_jobs import AsyncJobFilter
 from models_library.functions import FunctionJobCollection, FunctionJobID
 from models_library.products import ProductName
 from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
 from models_library.users import UserID
-from servicelib.celery.models import TaskFilter, TaskID, TaskMetadata, TasksQueue
+from servicelib.celery.models import TaskID, TaskMetadata, TasksQueue
 from servicelib.fastapi.dependencies import get_reverse_url_mapper
 from servicelib.utils import limited_gather
 
@@ -33,6 +32,7 @@
 from ...celery_worker.worker_tasks.functions_tasks import (
     run_function as run_function_task,
 )
+from ...clients.celery_task_manager import get_task_filter
 from ...exceptions.function_errors import FunctionJobCacheNotFoundError
 from ...models.pagination import Page, PaginationParams
 from ...models.schemas.errors import ErrorGet
@@ -44,7 +44,7 @@
     get_current_user_id,
     get_product_name,
 )
-from ..dependencies.celery import ASYNC_JOB_CLIENT_NAME, get_task_manager
+from ..dependencies.celery import get_task_manager
 from ..dependencies.services import get_function_job_service, get_function_service
 from ..dependencies.webserver_rpc import get_wb_api_rpc_client
 from ._constants import (
@@ -368,12 +368,9 @@ async def run_function(
     )
 
     # run function in celery task
-    job_filter = AsyncJobFilter(
-        user_id=user_identity.user_id,
-        product_name=user_identity.product_name,
-        client_name=ASYNC_JOB_CLIENT_NAME,
+    task_filter = get_task_filter(
+        user_id=user_identity.user_id, product_name=user_identity.product_name
     )
-    task_filter = TaskFilter.model_validate(job_filter.model_dump())
     task_name = run_function_task.__name__
 
     task_uuid = await task_manager.submit_task(

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

@@ -10,19 +10,18 @@
     TaskStatus,
 )
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
-    AsyncJobFilter,
     AsyncJobId,
 )
 from models_library.products import ProductName
 from models_library.users import UserID
-from servicelib.celery.models import TaskFilter, TaskState, TaskUUID
+from servicelib.celery.models import TaskState, TaskUUID
 from servicelib.fastapi.dependencies import get_app
 from servicelib.logging_errors import create_troubleshootting_log_kwargs
 
 from ...models.schemas.base import ApiServerEnvelope
 from ...models.schemas.errors import ErrorGet
 from ..dependencies.authentication import get_current_user_id, get_product_name
-from ..dependencies.celery import ASYNC_JOB_CLIENT_NAME, get_task_manager
+from ..dependencies.celery import get_task_manager
 from ._constants import (
     FMSG_CHANGELOG_NEW_IN_VERSION,
     create_route_description,
@@ -32,13 +31,6 @@
 _logger = logging.getLogger(__name__)
 
 
-def _get_task_filter(user_id: UserID, product_name: ProductName) -> TaskFilter:
-    job_filter = AsyncJobFilter(
-        user_id=user_id, product_name=product_name, client_name=ASYNC_JOB_CLIENT_NAME
-    )
-    return TaskFilter.model_validate(job_filter.model_dump())
-
-
 _DEFAULT_TASK_STATUS_CODES: dict[int | str, dict[str, Any]] = {
     status.HTTP_500_INTERNAL_SERVER_ERROR: {
         "description": "Internal server error",

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

@@ -1,11 +1,27 @@
 from celery_library.common import create_app, create_task_manager
 from celery_library.types import register_celery_types, register_pydantic_types
 from fastapi import FastAPI
+from models_library.api_schemas_rpc_async_jobs.async_jobs import AsyncJobFilter
+from models_library.products import ProductName
+from models_library.users import UserID
+from servicelib.celery.models import TaskFilter
 from settings_library.celery import CelerySettings
 
+from .._meta import APP_NAME
 from ..celery_worker.worker_tasks.tasks import pydantic_types_to_register
 
 
+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
+    )
+
+
+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())
+
+
 def setup_task_manager(app: FastAPI, celery_settings: CelerySettings) -> None:
     async def on_startup() -> None:
         app.state.task_manager = await create_task_manager(

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

@@ -2,7 +2,6 @@
 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
@@ -11,7 +10,7 @@
 from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
 from servicelib.rabbitmq.rpc_interfaces.storage import simcore_s3 as storage_rpc
 
-from ..api.dependencies.celery import ASYNC_JOB_CLIENT_NAME
+from ..clients.celery_task_manager import get_job_filter
 from ..exceptions.service_errors_utils import service_exception_mapper
 
 _exception_mapper = partial(service_exception_mapper, service_name="Storage")
@@ -32,10 +31,9 @@ async def start_data_export(
             self._rpc_client,
             paths_to_export=paths_to_export,
             export_as="download_link",
-            job_filter=AsyncJobFilter(
+            job_filter=get_job_filter(
                 user_id=self._user_id,
                 product_name=self._product_name,
-                client_name=ASYNC_JOB_CLIENT_NAME,
             ),
         )
         return async_job_get

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

@@ -23,7 +23,6 @@
 from fastapi import FastAPI, status
 from httpx import AsyncClient, BasicAuth, HTTPStatusError
 from models_library.api_schemas_long_running_tasks.tasks import TaskResult, TaskStatus
-from models_library.api_schemas_rpc_async_jobs.async_jobs import AsyncJobFilter
 from models_library.functions import (
     FunctionClass,
     FunctionID,
@@ -50,12 +49,12 @@
 from simcore_service_api_server._meta import API_VTAG
 from simcore_service_api_server.api.dependencies.authentication import Identity
 from simcore_service_api_server.api.dependencies.celery import (
-    ASYNC_JOB_CLIENT_NAME,
     get_task_manager,
 )
 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 (
@@ -281,10 +280,9 @@ async def test_celery_error_propagation(
     with_api_server_celery_worker: TestWorkController,
 ):
 
-    job_filter = AsyncJobFilter(
+    job_filter = get_job_filter(
         user_id=user_identity.user_id,
         product_name=user_identity.product_name,
-        client_name=ASYNC_JOB_CLIENT_NAME,
     )
     task_manager = get_task_manager(app=app)
     task_uuid = await task_manager.submit_task(