✨ JobService: Implement job listing with pagination and refactor SolverService and StudiesService to utilize it

pcrespov · pcrespov · commit 5d74010569f2 · 2025-04-29T19:46:33.000+02:00
diff --git a/services/api-server/src/simcore_service_api_server/_service_job.py b/services/api-server/src/simcore_service_api_server/_service_job.py
@@ -1,40 +1,33 @@
 import logging
 from collections.abc import Callable
-from typing import Annotated
 
-from fastapi import Depends
 from models_library.api_schemas_webserver.projects import ProjectCreateNew, ProjectGet
 from models_library.products import ProductName
 from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
+from models_library.rest_pagination import PageMetaInfoLimitOffset, PageOffsetInt
+from models_library.rpc_pagination import PageLimitInt
 from models_library.users import UserID
 from pydantic import HttpUrl
-from servicelib.fastapi.app_state import SingletonInAppStateMixin
 from servicelib.logging_utils import log_context
 
-from .api.dependencies.authentication import (
-    get_current_user_id,
-    get_product_name,
-)
-from .api.dependencies.webserver_http import get_webserver_session
-from .api.dependencies.webserver_rpc import (
-    get_wb_api_rpc_client,
-)
 from .models.schemas.jobs import Job, JobInputs
 from .models.schemas.programs import Program
 from .models.schemas.solvers import Solver
 from .services_http.solver_job_models_converters import (
     create_job_from_project,
+    create_job_inputs_from_node_inputs,
     create_new_project_for_job,
 )
 from .services_http.webserver import AuthSession
 from .services_rpc.wb_api_server import WbApiRpcClient
 
 _logger = logging.getLogger(__name__)
 
+DEFAULT_PAGINATION_LIMIT = 999  # MAXIMUM_NUMBER_OF_ITEMS_PER_PAGE - 1
 
-class JobService(SingletonInAppStateMixin):
-    app_state_name = "JobService"
+
+class JobService:
     _web_rest_api: AuthSession
     _web_rpc_api: WbApiRpcClient
     _user_id: UserID
@@ -43,16 +36,64 @@ class JobService(SingletonInAppStateMixin):
     def __init__(
         self,
         *,
-        web_rest_api: Annotated[AuthSession, Depends(get_webserver_session)],
-        web_rpc_api: Annotated[WbApiRpcClient, Depends(get_wb_api_rpc_client)],
-        user_id: Annotated[UserID, Depends(get_current_user_id)],
-        product_name: Annotated[ProductName, Depends(get_product_name)],
+        web_rest_api: AuthSession,
+        web_rpc_api: WbApiRpcClient,
+        user_id: UserID,
+        product_name: ProductName,
     ):
         self._web_rest_api = web_rest_api
         self._web_rpc_api = web_rpc_api
         self._user_id = user_id
         self._product_name = product_name
 
+    async def list_jobs_by_resource_prefix(
+        self,
+        *,
+        job_parent_resource_name_prefix: str,
+        offset: PageOffsetInt = 0,
+        limit: PageLimitInt = DEFAULT_PAGINATION_LIMIT,
+    ) -> tuple[list[Job], PageMetaInfoLimitOffset]:
+        """Lists all jobs for a user with pagination based on resource name prefix"""
+
+        # List projects marked as jobs
+        projects_page = await self._web_rpc_api.list_projects_marked_as_jobs(
+            product_name=self._product_name,
+            user_id=self._user_id,
+            offset=offset,
+            limit=limit,
+            job_parent_resource_name_prefix=job_parent_resource_name_prefix,
+        )
+
+        # Convert projects to jobs
+        jobs: list[Job] = []
+        for project_job in projects_page.data:
+            assert (  # nosec
+                len(project_job.workbench) == 1
+            ), "Expected only one solver node in workbench"
+
+            solver_node = next(iter(project_job.workbench.values()))
+            job_inputs: JobInputs = create_job_inputs_from_node_inputs(
+                inputs=solver_node.inputs or {}
+            )
+            assert project_job.job_parent_resource_name  # nosec
+
+            jobs.append(
+                Job(
+                    id=project_job.uuid,
+                    name=Job.compose_resource_name(
+                        project_job.job_parent_resource_name, project_job.uuid
+                    ),
+                    inputs_checksum=job_inputs.compute_checksum(),
+                    created_at=project_job.created_at,
+                    runner_name=project_job.job_parent_resource_name,
+                    url=None,
+                    runner_url=None,
+                    outputs_url=None,
+                )
+            )
+
+        return jobs, projects_page.meta
+
     async def create_job(
         self,
         *,
diff --git a/services/api-server/src/simcore_service_api_server/_service_solvers.py b/services/api-server/src/simcore_service_api_server/_service_solvers.py
@@ -2,7 +2,6 @@
 from models_library.api_schemas_catalog.services import ServiceListFilters
 from models_library.basic_types import VersionStr
 from models_library.products import ProductName
-from models_library.projects_nodes import Node
 from models_library.rest_pagination import (
     MAXIMUM_NUMBER_OF_ITEMS_PER_PAGE,
     PageMetaInfoLimitOffset,
@@ -18,29 +17,26 @@
     SolverServiceListJobsFiltersError,
 )
 
+from ._service_job import JobService
 from .models.api_resources import compose_resource_name
-from .models.schemas.jobs import Job, JobInputs
+from .models.schemas.jobs import Job
 from .models.schemas.solvers import Solver, SolverKeyId
-from .services_http.solver_job_models_converters import (
-    create_job_inputs_from_node_inputs,
-)
 from .services_rpc.catalog import CatalogService
-from .services_rpc.wb_api_server import WbApiRpcClient
 
 DEFAULT_PAGINATION_LIMIT = MAXIMUM_NUMBER_OF_ITEMS_PER_PAGE - 1
 
 
 class SolverService:
     _catalog_service: CatalogService
-    _webserver_client: WbApiRpcClient
+    _job_service: JobService
 
     def __init__(
         self,
         catalog_service: CatalogService,
-        webserver_client: WbApiRpcClient,
+        job_service: JobService,
     ):
         self._catalog_service = catalog_service
-        self._webserver_client = webserver_client
+        self._job_service = job_service
 
     async def get_solver(
         self,
@@ -94,8 +90,6 @@ async def get_latest_release(
     async def list_jobs(
         self,
         *,
-        product_name: ProductName,
-        user_id: UserID,
         # filters
         solver_key: SolverKeyId | None = None,
         solver_version: VersionStr | None = None,
@@ -121,46 +115,13 @@ async def list_jobs(
             *collection_or_resource_ids
         )
 
-        # 2. List projects marked as jobs
-        projects_page = await self._webserver_client.list_projects_marked_as_jobs(
-            product_name=product_name,
-            user_id=user_id,
+        # Use the common implementation from JobService
+        return await self._job_service.list_jobs_by_resource_prefix(
             offset=offset,
             limit=limit,
             job_parent_resource_name_prefix=job_parent_resource_name_prefix,
         )
 
-        # 3. Convert projects to jobs
-        jobs: list[Job] = []
-        for project_job in projects_page.data:
-
-            assert (  # nosec
-                len(project_job.workbench) == 1
-            ), "Expected only one solver node in workbench"
-
-            solver_node: Node = next(iter(project_job.workbench.values()))
-            job_inputs: JobInputs = create_job_inputs_from_node_inputs(
-                inputs=solver_node.inputs or {}
-            )
-            assert project_job.job_parent_resource_name  # nosec
-
-            jobs.append(
-                Job(
-                    id=project_job.uuid,
-                    name=Job.compose_resource_name(
-                        project_job.job_parent_resource_name, project_job.uuid
-                    ),
-                    inputs_checksum=job_inputs.compute_checksum(),
-                    created_at=project_job.created_at,
-                    runner_name=project_job.job_parent_resource_name,
-                    url=None,
-                    runner_url=None,
-                    outputs_url=None,
-                )
-            )
-
-        return jobs, projects_page.meta
-
     async def solver_release_history(
         self,
         *,
diff --git a/services/api-server/src/simcore_service_api_server/_service_studies.py b/services/api-server/src/simcore_service_api_server/_service_studies.py
@@ -1,38 +1,30 @@
-from models_library.products import ProductName
-from models_library.projects_nodes import Node
 from models_library.rest_pagination import (
     MAXIMUM_NUMBER_OF_ITEMS_PER_PAGE,
     PageMetaInfoLimitOffset,
     PageOffsetInt,
 )
 from models_library.rpc_pagination import PageLimitInt
-from models_library.users import UserID
 from simcore_service_api_server.models.schemas.studies import StudyID
 
+from ._service_job import JobService
 from .models.api_resources import compose_resource_name
-from .models.schemas.jobs import Job, JobInputs
-from .services_http.solver_job_models_converters import (
-    create_job_inputs_from_node_inputs,
-)
-from .services_rpc.wb_api_server import WbApiRpcClient
+from .models.schemas.jobs import Job
 
 DEFAULT_PAGINATION_LIMIT = MAXIMUM_NUMBER_OF_ITEMS_PER_PAGE - 1
 
 
 class StudiesService:
-    _webserver_client: WbApiRpcClient
+    _job_service: JobService
 
     def __init__(
         self,
-        webserver_client: WbApiRpcClient,
+        job_service: JobService,
     ):
-        self._webserver_client = webserver_client
+        self._job_service = job_service
 
     async def list_jobs(
         self,
         *,
-        product_name: ProductName,
-        user_id: UserID,
         # filters
         study_id: StudyID | None = None,
         # pagination
@@ -52,42 +44,9 @@ async def list_jobs(
             *collection_or_resource_ids
         )
 
-        # 2. List projects marked as jobs
-        projects_page = await self._webserver_client.list_projects_marked_as_jobs(
-            product_name=product_name,
-            user_id=user_id,
+        # Use the common implementation from JobService
+        return await self._job_service.list_jobs_by_resource_prefix(
             offset=offset,
             limit=limit,
             job_parent_resource_name_prefix=job_parent_resource_name_prefix,
         )
-
-        # 3. Convert projects to jobs
-        jobs: list[Job] = []
-        for project_job in projects_page.data:
-
-            assert (  # nosec
-                len(project_job.workbench) == 1
-            ), "Expected only one solver node in workbench"
-
-            solver_node: Node = next(iter(project_job.workbench.values()))
-            job_inputs: JobInputs = create_job_inputs_from_node_inputs(
-                inputs=solver_node.inputs or {}
-            )
-            assert project_job.job_parent_resource_name  # nosec
-
-            jobs.append(
-                Job(
-                    id=project_job.uuid,
-                    name=Job.compose_resource_name(
-                        project_job.job_parent_resource_name, project_job.uuid
-                    ),
-                    inputs_checksum=job_inputs.compute_checksum(),
-                    created_at=project_job.created_at,
-                    runner_name=project_job.job_parent_resource_name,
-                    url=None,
-                    runner_url=None,
-                    outputs_url=None,
-                )
-            )
-
-        return jobs, projects_page.meta
diff --git a/services/api-server/src/simcore_service_api_server/api/dependencies/services.py b/services/api-server/src/simcore_service_api_server/api/dependencies/services.py
@@ -4,14 +4,19 @@
 from typing import Annotated
 
 from fastapi import Depends, HTTPException, Request, status
+from models_library.products import ProductName
+from models_library.users import UserID
 from servicelib.rabbitmq import RabbitMQRPCClient
 from simcore_service_api_server._service_studies import StudiesService
 
+from ..._service_job import JobService
 from ..._service_solvers import SolverService
 from ...services_rpc.catalog import CatalogService
 from ...services_rpc.wb_api_server import WbApiRpcClient
 from ...utils.client_base import BaseServiceClientApi
+from .authentication import get_current_user_id, get_product_name
 from .rabbitmq import get_rabbitmq_rpc_client
+from .webserver_http import AuthSession, get_webserver_session
 from .webserver_rpc import get_wb_api_rpc_client
 
 
@@ -50,9 +55,27 @@ def get_catalog_service(
     return CatalogService(client=rpc_client)
 
 
+def get_job_service(
+    web_rest_api: Annotated[AuthSession, Depends(get_webserver_session)],
+    web_rpc_api: Annotated[WbApiRpcClient, Depends(get_wb_api_rpc_client)],
+    user_id: Annotated[UserID, Depends(get_current_user_id)],
+    product_name: Annotated[ProductName, Depends(get_product_name)],
+) -> JobService:
+    """
+    "Assembles" the JobsService layer to the underlying service and client interfaces
+    in the context of the rest controller (i.e. api/dependencies)
+    """
+    return JobService(
+        web_rest_api=web_rest_api,
+        web_rpc_api=web_rpc_api,
+        user_id=user_id,
+        product_name=product_name,
+    )
+
+
 def get_solver_service(
     catalog_service: Annotated[CatalogService, Depends(get_catalog_service)],
-    webserver_client: Annotated[WbApiRpcClient, Depends(get_wb_api_rpc_client)],
+    job_service: Annotated[JobService, Depends(get_job_service)],
 ) -> SolverService:
     """
     "Assembles" the SolverService layer to the underlying service and client interfaces
@@ -61,13 +84,13 @@ def get_solver_service(
 
     return SolverService(
         catalog_service=catalog_service,
-        webserver_client=webserver_client,
+        job_service=job_service,
     )
 
 
 def get_studies_service(
-    webserver_client: Annotated[WbApiRpcClient, Depends(get_wb_api_rpc_client)],
+    job_service: Annotated[JobService, Depends(get_job_service)],
 ) -> StudiesService:
     return StudiesService(
-        webserver_client=webserver_client,
+        job_service=job_service,
     )
