cleanup

Author: bisgaard-itis

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

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import NamedTuple, overload
+from typing import overload
 
 import jsonschema
 from common_library.exclude import as_dict_exclude_none
@@ -42,18 +42,14 @@
 
 from ._service_jobs import JobService
 from .models.api_resources import JobLinks
+from .models.domain.functions import PreRegisteredFunctionJobData
 from .models.schemas.jobs import (
     JobInputs,
     JobPricingSpecification,
 )
 from .services_rpc.wb_api_server import WbApiRpcClient
 
 
-class RegisteredFunctionJobData(NamedTuple):
-    function_job_id: FunctionJobID
-    job_inputs: JobInputs
-
-
 def join_inputs(
     default_inputs: FunctionInputs | None,
     function_inputs: FunctionInputs | None,
@@ -191,10 +187,15 @@ async def get_cached_function_job(
         self,
         *,
         function: RegisteredFunction,
-        function_inputs: FunctionInputs,
         job_inputs: JobInputs,
     ) -> RegisteredFunctionJob:
-        """raises FunctionJobCacheNotFoundError if no cached job is found"""
+        """
+        N.B. this function checks access rights
+
+        raises FunctionsExecuteApiAccessDeniedError if user cannot execute functions
+        raises FunctionJobCacheNotFoundError if no cached job is found
+
+        """
 
         user_api_access_rights = (
             await self._web_rpc_client.get_functions_user_api_access_rights(
@@ -234,13 +235,12 @@ async def get_cached_function_job(
 
         raise FunctionJobCacheNotFoundError()
 
-    async def create_registered_function_job(
+    async def pre_register_function_job(
         self,
         *,
         function: RegisteredFunction,
-        function_inputs: FunctionInputs,
         job_inputs: JobInputs,
-    ) -> FunctionJobID:
+    ) -> PreRegisteredFunctionJobData:
 
         if function.input_schema is not None:
             is_valid, validation_str = await self.validate_function_inputs(
@@ -284,7 +284,10 @@ async def create_registered_function_job(
                 function_class=function.function_class,
             )
 
-        return job.uid
+        return PreRegisteredFunctionJobData(
+            function_job_id=job.uid,
+            job_inputs=job_inputs,
+        )
 
     @overload
     async def patch_registered_function_job(
@@ -368,8 +371,7 @@ async def run_function(
         *,
         job_creation_task_id: TaskID | None,
         function: RegisteredFunction,
-        pre_registered_function_job_id: FunctionJobID,
-        job_inputs: JobInputs,
+        pre_registered_function_job_data: PreRegisteredFunctionJobData,
         pricing_spec: JobPricingSpecification | None,
         job_links: JobLinks,
         x_simcore_parent_project_uuid: NodeID | None,
@@ -380,7 +382,7 @@ async def run_function(
         if function.function_class == FunctionClass.PROJECT:
             study_job = await self._job_service.create_studies_job(
                 study_id=function.project_id,
-                job_inputs=job_inputs,
+                job_inputs=pre_registered_function_job_data.job_inputs,
                 hidden=True,
                 job_links=job_links,
                 x_simcore_parent_project_uuid=x_simcore_parent_project_uuid,
@@ -394,7 +396,7 @@ async def run_function(
             return await self.patch_registered_function_job(
                 user_id=self.user_id,
                 product_name=self.product_name,
-                function_job_id=pre_registered_function_job_id,
+                function_job_id=pre_registered_function_job_data.function_job_id,
                 function_class=FunctionClass.PROJECT,
                 job_creation_task_id=job_creation_task_id,
                 project_job_id=study_job.id,
@@ -404,7 +406,7 @@ async def run_function(
             solver_job = await self._job_service.create_solver_job(
                 solver_key=function.solver_key,
                 version=function.solver_version,
-                inputs=job_inputs,
+                inputs=pre_registered_function_job_data.job_inputs,
                 job_links=job_links,
                 hidden=True,
                 x_simcore_parent_project_uuid=x_simcore_parent_project_uuid,
@@ -419,7 +421,7 @@ async def run_function(
             return await self.patch_registered_function_job(
                 user_id=self.user_id,
                 product_name=self.product_name,
-                function_job_id=pre_registered_function_job_id,
+                function_job_id=pre_registered_function_job_data.function_job_id,
                 function_class=FunctionClass.SOLVER,
                 job_creation_task_id=job_creation_task_id,
                 solver_job_id=solver_job.id,
@@ -434,7 +436,7 @@ async def map_function(
         *,
         job_creation_task_id: TaskID | None,
         function: RegisteredFunction,
-        job_inputs: list[JobInputs],
+        pre_registered_function_job_data_list: list[PreRegisteredFunctionJobData],
         job_links: JobLinks,
         pricing_spec: JobPricingSpecification | None,
         x_simcore_parent_project_uuid: ProjectID | None,
@@ -445,16 +447,16 @@ async def map_function(
             await self.run_function(
                 job_creation_task_id=job_creation_task_id,
                 function=function,
-                job_inputs=inputs,
+                pre_registered_function_job_data=data,
                 pricing_spec=pricing_spec,
                 job_links=job_links,
                 x_simcore_parent_project_uuid=x_simcore_parent_project_uuid,
                 x_simcore_parent_node_id=x_simcore_parent_node_id,
             )
-            for inputs in job_inputs
+            for data in pre_registered_function_job_data_list
         ]
 
-        function_job_collection_description = f"Function job collection of map of function {function.uid} with {len(job_inputs)} inputs"
+        function_job_collection_description = f"Function job collection of map of function {function.uid} with {len(pre_registered_function_job_data_list)} inputs"
         return await self._web_rpc_client.register_function_job_collection(
             function_job_collection=FunctionJobCollection(
                 title="Function job collection of function map",

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

@@ -334,6 +334,7 @@ async def run_function(  # noqa: PLR0913
     x_simcore_parent_project_uuid: Annotated[ProjectID | Literal["null"], Header()],
     x_simcore_parent_node_id: Annotated[NodeID | Literal["null"], Header()],
 ) -> RegisteredFunctionJob:
+    # massage inputs
     task_manager = get_task_manager(request.app)
     parent_project_uuid = (
         x_simcore_parent_project_uuid
@@ -351,20 +352,18 @@ async def run_function(  # noqa: PLR0913
         function=to_run_function, function_inputs=function_inputs
     )
 
+    # check if results are cached
     try:
-        # checks access rights
         return await function_job_service.get_cached_function_job(
             function=to_run_function,
-            function_inputs=function_inputs,
             job_inputs=job_inputs,
         )
     except FunctionJobCacheNotFoundError:
         pass
 
-    pre_registered_function_job_id = (
-        await function_job_service.create_registered_function_job(
+    pre_registered_function_job_data = (
+        await function_job_service.pre_register_function_job(
             function=to_run_function,
-            function_inputs=function_inputs,
             job_inputs=job_inputs,
         )
     )
@@ -387,8 +386,7 @@ async def run_function(  # noqa: PLR0913
         task_filter=task_filter,
         user_identity=user_identity,
         function=to_run_function,
-        pre_registered_function_job_id=pre_registered_function_job_id,
-        job_inputs=job_inputs,
+        pre_registered_function_job_data=pre_registered_function_job_data,
         pricing_spec=pricing_spec,
         job_links=job_links,
         x_simcore_parent_project_uuid=parent_project_uuid,
@@ -398,7 +396,7 @@ async def run_function(  # noqa: PLR0913
     return await function_job_service.patch_registered_function_job(
         user_id=user_identity.user_id,
         product_name=user_identity.product_name,
-        function_job_id=pre_registered_function_job_id,
+        function_job_id=pre_registered_function_job_data.function_job_id,
         function_class=to_run_function.function_class,
         job_creation_task_id=TaskID(task_uuid),
     )

services/api-server/src/simcore_service_api_server/celery/worker_tasks/functions_tasks.py

@@ -3,7 +3,7 @@
 )
 from celery_library.utils import get_app_server  # pylint: disable=no-name-in-module
 from fastapi import FastAPI
-from models_library.functions import FunctionJobID, RegisteredFunction
+from models_library.functions import RegisteredFunction
 from models_library.projects_nodes_io import NodeID
 from servicelib.celery.models import TaskID
 from simcore_service_api_server._service_function_jobs import FunctionJobService
@@ -21,7 +21,8 @@
 from ...api.dependencies.webserver_http import get_session_cookie, get_webserver_session
 from ...api.dependencies.webserver_rpc import get_wb_api_rpc_client
 from ...models.api_resources import JobLinks
-from ...models.schemas.jobs import JobInputs, JobPricingSpecification
+from ...models.domain.functions import PreRegisteredFunctionJobData
+from ...models.schemas.jobs import JobPricingSpecification
 from ...services_http.director_v2 import DirectorV2Api
 from ...services_http.storage import StorageApi
 
@@ -94,8 +95,7 @@ async def run_function(
     *,
     user_identity: Identity,
     function: RegisteredFunction,
-    pre_registered_function_job_id: FunctionJobID,
-    job_inputs: JobInputs,
+    pre_registered_function_job_data: PreRegisteredFunctionJobData,
     pricing_spec: JobPricingSpecification | None,
     job_links: JobLinks,
     x_simcore_parent_project_uuid: NodeID | None,
@@ -110,8 +110,7 @@ async def run_function(
     return await function_job_service.run_function(
         job_creation_task_id=task_id,
         function=function,
-        pre_registered_function_job_id=pre_registered_function_job_id,
-        job_inputs=job_inputs,
+        pre_registered_function_job_data=pre_registered_function_job_data,
         pricing_spec=pricing_spec,
         job_links=job_links,
         x_simcore_parent_project_uuid=x_simcore_parent_project_uuid,

services/api-server/src/simcore_service_api_server/models/domain/functions.py

@@ -0,0 +1,9 @@
+from models_library.functions import FunctionJobID
+from pydantic import BaseModel
+
+from ...models.schemas.jobs import JobInputs
+
+
+class PreRegisteredFunctionJobData(BaseModel):
+    function_job_id: FunctionJobID
+    job_inputs: JobInputs