Finish merge with celery worker PR

Author: wvangeit

.github/copilot-instructions.md

@@ -41,9 +41,9 @@ async def ports(
 
 __all__ = (
     "DBManager",
-    "exceptions",
     "FileLinkType",
     "Nodeports",
     "Port",
+    "exceptions",
     "ports",
 )

packages/simcore-sdk/src/simcore_sdk/node_ports_v2/__init__.py

@@ -136,8 +136,14 @@ info: ## displays service info
 #
 
 .PHONY: _run-test-dev _run-test-ci
+# Use OVERRIDE_TEST_TARGET to override the default test target
+# Use TEST_DISABLE_PDB to disable pdb in these tests
+TEST_TARGET := $(if $(OVERRIDE_TEST_TARGET), $(OVERRIDE_TEST_TARGET), $(if $(target),$(target),$(CURDIR)/tests/unit))
+# Use TEST_DISABLE_PDB to disable pdb in these tests
+PDB_OPTION := $(if $(filter-out 0,$(TEST_DISABLE_PDB)),,--pdb)
+# Use TEST_ENABLE_TESTIT to enable testit in these tests
+TESTIT_OPTION := $(if $(filter-out 0,$(TEST_ENABLE_TESTIT)),-m testit,)
 
-TEST_TARGET := $(if $(target),$(target),$(CURDIR)/tests/unit)
 PYTEST_ADDITIONAL_PARAMETERS := $(if $(pytest-parameters),$(pytest-parameters),)
 _run-test-dev: _check_venv_active
 	# runs tests for development (e.g w/ pdb)
@@ -153,7 +159,8 @@ _run-test-dev: _check_venv_active
 		--failed-first \
 		--junitxml=junit.xml -o junit_family=legacy \
 		--keep-docker-up \
-		--pdb \
+		$(PDB_OPTION) \
+		$(TESTIT_OPTION) \
 		-vv \
 		$(PYTEST_ADDITIONAL_PARAMETERS) \
 		$(TEST_TARGET)

scripts/common-service.Makefile

@@ -39,7 +39,7 @@
 from models_library.users import UserID
 from pydantic import ValidationError
 from simcore_service_api_server._service_functions import FunctionService
-from simcore_service_api_server.services_http.storage import StorageApi
+from simcore_service_api_server.services_rpc.storage import StorageService
 from sqlalchemy.ext.asyncio import AsyncEngine
 
 from ._service_jobs import JobService
@@ -73,11 +73,10 @@ class FunctionJobService:
     user_id: UserID
     product_name: ProductName
     _web_rpc_client: WbApiRpcClient
-    _storage_client: StorageApi
+    _storage_client: StorageService
     _job_service: JobService
     _function_service: FunctionService
     _webserver_api: AuthSession
-    _async_pg_engine: AsyncEngine
 
     async def list_function_jobs(
         self,
@@ -121,41 +120,15 @@ async def list_function_jobs_with_status(
             pagination_offset=pagination_offset, pagination_limit=pagination_limit
         )
 
-        function_jobs_wso, page_meta = (
-            await self._web_rpc_client.list_function_jobs_with_status(
-                user_id=self.user_id,
-                product_name=self.product_name,
-                filter_by_function_id=filter_by_function_id,
-                filter_by_function_job_ids=filter_by_function_job_ids,
-                filter_by_function_job_collection_id=filter_by_function_job_collection_id,
-                **pagination_kwargs,
-            )
+        return await self._web_rpc_client.list_function_jobs_with_status(
+            user_id=self.user_id,
+            product_name=self.product_name,
+            filter_by_function_id=filter_by_function_id,
+            filter_by_function_job_ids=filter_by_function_job_ids,
+            filter_by_function_job_collection_id=filter_by_function_job_collection_id,
+            **pagination_kwargs,
         )
 
-        for function_job_wso in function_jobs_wso:
-            if (
-                function_job_wso.status.status
-                not in (
-                    RunningState.SUCCESS,
-                    RunningState.FAILED,
-                )
-            ) or function_job_wso.outputs is None:
-                function = await self._function_service.get_function(
-                    function_id=function_job_wso.function_uid
-                )
-                function_job_wso.status = await self.inspect_function_job(
-                    function=function, function_job=function_job_wso
-                )
-                if function_job_wso.status.status == RunningState.SUCCESS:
-                    function_job_wso.outputs = await self.function_job_outputs(
-                        function_job=function_job_wso,
-                        function=function,
-                        user_id=self.user_id,
-                        product_name=self.product_name,
-                        stored_job_outputs=None,
-                    )
-        return function_jobs_wso, page_meta
-
     async def validate_function_inputs(
         self, *, function_id: FunctionID, inputs: FunctionInputs
     ) -> tuple[bool, str]:
@@ -525,15 +498,19 @@ async def function_job_outputs(
         user_id: UserID,
         product_name: ProductName,
         stored_job_outputs: FunctionOutputs | None,
+        async_pg_engine: AsyncEngine,
     ) -> FunctionOutputs:
 
         if stored_job_outputs is not None:
             return stored_job_outputs
 
-        job_status = await self.inspect_function_job(
-            function=function,
-            function_job=function_job,
-        )
+        try:
+            job_status = await self.inspect_function_job(
+                function=function,
+                function_job=function_job,
+            )
+        except FunctionJobProjectMissingError:
+            return None
 
         if job_status.status != RunningState.SUCCESS:
             return None
@@ -564,6 +541,7 @@ async def function_job_outputs(
                         solver_key=function.solver_key,
                         version=function.solver_version,
                         job_id=function_job.solver_job_id,
+                        async_pg_engine=async_pg_engine,
                     )
                 ).results
             )

services/api-server/src/simcore_service_api_server/_service_function_jobs.py

@@ -99,7 +99,6 @@ class JobService:
     _storage_rest_client: StorageApi
     _directorv2_rpc_client: DirectorV2Service
     _solver_service: SolverService
-    _async_pg_engine: AsyncEngine
 
     user_id: UserID
     product_name: ProductName
@@ -300,7 +299,11 @@ async def get_job(
         )
 
     async def get_solver_job_outputs(
-        self, solver_key: SolverKeyId, version: VersionStr, job_id: JobID
+        self,
+        solver_key: SolverKeyId,
+        version: VersionStr,
+        job_id: JobID,
+        async_pg_engine: AsyncEngine,
     ) -> JobOutputs:
         job_name = compose_solver_job_resource_name(solver_key, version, job_id)
         _logger.debug("Get Job '%s' outputs", job_name)
@@ -340,7 +343,7 @@ async def get_solver_job_outputs(
             user_id=self.user_id,
             project_uuid=job_id,
             node_uuid=UUID(node_ids[0]),
-            db_engine=self._async_pg_engine,
+            db_engine=async_pg_engine,
         )
 
         results: dict[str, ArgumentTypes] = {}

services/api-server/src/simcore_service_api_server/_service_jobs.py

@@ -7,7 +7,6 @@
 from models_library.products import ProductName
 from models_library.users import UserID
 from servicelib.rabbitmq import RabbitMQRPCClient
-from sqlalchemy.ext.asyncio import AsyncEngine
 
 from ..._service_function_jobs import FunctionJobService
 from ..._service_functions import FunctionService
@@ -22,7 +21,6 @@
 from ...services_rpc.storage import StorageService
 from ...services_rpc.wb_api_server import WbApiRpcClient
 from ...utils.client_base import BaseServiceClientApi
-from ..dependencies.database import get_db_asyncpg_engine
 from .authentication import get_current_user_id, get_product_name
 from .rabbitmq import get_rabbitmq_rpc_client
 from .webserver_http import get_webserver_session
@@ -120,7 +118,6 @@ def get_job_service(
     user_id: Annotated[UserID, Depends(get_current_user_id)],
     product_name: Annotated[ProductName, Depends(get_product_name)],
     solver_service: Annotated[SolverService, Depends(get_solver_service)],
-    async_pg_engine: Annotated[AsyncEngine, Depends(get_db_asyncpg_engine)],
 ) -> JobService:
     """
     "Assembles" the JobsService layer to the underlying service and client interfaces
@@ -134,7 +131,6 @@ def get_job_service(
         _director2_api=director2_api,
         _storage_rest_client=storage_api,
         _solver_service=solver_service,
-        _async_pg_engine=async_pg_engine,
         user_id=user_id,
         product_name=product_name,
     )
@@ -159,16 +155,15 @@ def get_function_job_service(
     user_id: Annotated[UserID, Depends(get_current_user_id)],
     product_name: Annotated[ProductName, Depends(get_product_name)],
     webserver_api: Annotated[AuthSession, Depends(get_webserver_session)],
-    storage_service: Annotated[StorageApi, Depends(get_storage_service)],
-    async_pg_engine: Annotated[AsyncEngine, Depends(get_db_asyncpg_engine)],
+    storage_service: Annotated[StorageService, Depends(get_storage_service)],
+    # async_pg_engine: Annotated[AsyncEngine, Depends(get_db_asyncpg_engine)],
 ) -> FunctionJobService:
     return FunctionJobService(
         _web_rpc_client=web_rpc_api,
         _job_service=job_service,
         _function_service=function_service,
         _storage_client=storage_service,
         _webserver_api=webserver_api,
-        _async_pg_engine=async_pg_engine,
         user_id=user_id,
         product_name=product_name,
     )

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

@@ -20,23 +20,24 @@
     UnsupportedFunctionFunctionJobClassCombinationError,
 )
 from models_library.products import ProductName
+from models_library.projects_state import RunningState
 from models_library.users import UserID
 from servicelib.celery.models import TaskUUID
 from servicelib.fastapi.dependencies import get_app
 from servicelib.logging_errors import create_troubleshootting_log_kwargs
-from simcore_service_api_server._service_functions import FunctionService
-from simcore_service_api_server.models.schemas.functions_filters import (
-    FunctionJobsListFilters,
-)
+from sqlalchemy.ext.asyncio import AsyncEngine
 
 from ..._service_function_jobs import FunctionJobService
+from ..._service_functions import FunctionService
 from ..._service_jobs import JobService
 from ...exceptions.function_errors import FunctionJobProjectMissingError
 from ...models.pagination import Page, PaginationParams
 from ...models.schemas.errors import ErrorGet
+from ...models.schemas.functions_filters import FunctionJobsListFilters
 from ...services_rpc.wb_api_server import WbApiRpcClient
 from ..dependencies.authentication import get_current_user_id, get_product_name
 from ..dependencies.celery import get_task_manager
+from ..dependencies.database import get_db_asyncpg_engine
 from ..dependencies.functions import (
     get_function_from_functionjob,
     get_function_job_dependency,
@@ -104,11 +105,11 @@
             )
         )
 
-    if endpoint in ["list_function_jobs_with_status"]:
+    if endpoint in ["list_function_jobs"]:
         CHANGE_LOGS[endpoint].append(
             FMSG_CHANGELOG_ADDED_IN_VERSION.format(
                 WITH_STATUS_RELEASE_VERSION,
-                "add /with-status endpoint to list function jobs with their status",
+                "add include_status bool query parameter to list function jobs with their status",
             )
         )
 
@@ -123,11 +124,16 @@
     ),
 )
 async def list_function_jobs(
+    app: Annotated[FastAPI, Depends(get_app)],
     page_params: Annotated[PaginationParams, Depends()],
     function_job_service: Annotated[
         FunctionJobService, Depends(get_function_job_service)
     ],
+    function_service: Annotated[FunctionService, Depends(get_function_service)],
     filters: Annotated[FunctionJobsListFilters, Depends(get_function_jobs_filters)],
+    async_pg_engine: Annotated[AsyncEngine, Depends(get_db_asyncpg_engine)],
+    user_id: Annotated[UserID, Depends(get_current_user_id)],
+    product_name: Annotated[ProductName, Depends(get_product_name)],
     include_status: Annotated[  # noqa: FBT002
         bool, Query(description="Include job status in response")
     ] = False,
@@ -145,6 +151,38 @@ async def list_function_jobs(
                 filter_by_function_id=filters.function_id,
             )
         )
+        # the code below should ideally be in the service layer, but this can only be done if the
+        # celery status resolution is done in the service layer too
+        for function_job_wso in function_jobs_list:
+            if (
+                function_job_wso.status.status
+                not in (
+                    RunningState.SUCCESS,
+                    RunningState.FAILED,
+                )
+            ) or function_job_wso.outputs is None:
+                function = await function_service.get_function(
+                    function_id=function_job_wso.function_uid
+                )
+                function_job_wso.status = await function_job_status(
+                    app=app,
+                    function=function,
+                    function_job=function_job_wso,
+                    function_job_service=function_job_service,
+                    user_id=user_id,
+                    product_name=product_name,
+                )
+                if function_job_wso.status.status == RunningState.SUCCESS:
+                    function_job_wso.outputs = (
+                        await function_job_service.function_job_outputs(
+                            function_job=function_job_wso,
+                            function=function,
+                            user_id=user_id,
+                            product_name=product_name,
+                            stored_job_outputs=None,
+                            async_pg_engine=async_pg_engine,
+                        )
+                    )
     else:
         function_jobs_list, meta = await function_job_service.list_function_jobs(
             pagination_offset=page_params.offset,
@@ -322,13 +360,15 @@ async def function_job_outputs(
     user_id: Annotated[UserID, Depends(get_current_user_id)],
     product_name: Annotated[ProductName, Depends(get_product_name)],
     stored_job_outputs: Annotated[FunctionOutputs, Depends(get_stored_job_outputs)],
+    async_pg_engine: Annotated[AsyncEngine, Depends(get_db_asyncpg_engine)],
 ) -> FunctionOutputs:
     return await function_job_service.function_job_outputs(
         function_job=function_job,
         user_id=user_id,
         product_name=product_name,
         function=function,
         stored_job_outputs=stored_job_outputs,
+        async_pg_engine=async_pg_engine,
     )