implement run function workflow

Author: bisgaard-itis

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

@@ -286,6 +286,17 @@ async def create_registered_function_job(
 
         return job.uid
 
+    @overload
+    async def patch_registered_function_job(
+        self,
+        *,
+        user_id: UserID,
+        product_name: ProductName,
+        function_job_id: FunctionJobID,
+        function_class: FunctionClass,
+        job_creation_task_id: TaskID | None,
+    ) -> RegisteredFunctionJob: ...
+
     @overload
     async def patch_registered_function_job(
         self,
@@ -354,6 +365,7 @@ async def patch_registered_function_job(
     async def run_function(
         self,
         *,
+        job_creation_task_id: TaskID | None,
         function: RegisteredFunction,
         job_inputs: JobInputs,
         pricing_spec: JobPricingSpecification | None,
@@ -377,18 +389,13 @@ async def run_function(
                 job_id=study_job.id,
                 pricing_spec=pricing_spec,
             )
-            return await self._web_rpc_client.register_function_job(
-                function_job=ProjectFunctionJob(
-                    function_uid=function.uid,
-                    title=f"Function job of function {function.uid}",
-                    description=function.description,
-                    inputs=job_inputs.values,
-                    outputs=None,
-                    project_job_id=study_job.id,
-                    job_creation_task_id=None,
-                ),
+            return await self.patch_registered_function_job(
                 user_id=self.user_id,
                 product_name=self.product_name,
+                function_job_id=study_job.id,
+                function_class=FunctionClass.PROJECT,
+                job_creation_task_id=job_creation_task_id,
+                project_job_id=study_job.id,
             )
 
         if function.function_class == FunctionClass.SOLVER:
@@ -407,18 +414,13 @@ async def run_function(
                 job_id=solver_job.id,
                 pricing_spec=pricing_spec,
             )
-            return await self._web_rpc_client.register_function_job(
-                function_job=SolverFunctionJob(
-                    function_uid=function.uid,
-                    title=f"Function job of function {function.uid}",
-                    description=function.description,
-                    inputs=job_inputs.values,
-                    outputs=None,
-                    solver_job_id=solver_job.id,
-                    job_creation_task_id=None,
-                ),
+            return await self.patch_registered_function_job(
                 user_id=self.user_id,
                 product_name=self.product_name,
+                function_job_id=solver_job.id,
+                function_class=FunctionClass.SOLVER,
+                job_creation_task_id=job_creation_task_id,
+                solver_job_id=solver_job.id,
             )
 
         raise UnsupportedFunctionClassError(

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

@@ -17,12 +17,13 @@
     RegisteredFunctionJobCollection,
 )
 from models_library.api_schemas_rpc_async_jobs.async_jobs import AsyncJobFilter
+from models_library.functions import FunctionClass
 from models_library.functions_errors import FunctionJobCacheNotFoundError
 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, TaskMetadata, TasksQueue
+from servicelib.celery.models import TaskFilter, TaskID, TaskMetadata, TasksQueue
 from servicelib.fastapi.dependencies import get_reverse_url_mapper
 
 from ..._service_function_jobs import FunctionJobService
@@ -347,10 +348,10 @@ async def run_function(  # noqa: PLR0913
     )
     pricing_spec = JobPricingSpecification.create_from_headers(request.headers)
     job_links = await function_service.get_function_job_links(to_run_function, url_for)
-
     job_inputs = await function_job_service.create_function_job_inputs(
         function=to_run_function, function_inputs=function_inputs
     )
+
     try:
         # checks access rights
         return await function_job_service.get_cached_function_job(
@@ -398,10 +399,8 @@ async def run_function(  # noqa: PLR0913
         user_id=user_identity.user_id,
         product_name=user_identity.product_name,
         function_job_id=pre_registered_function_job_id,
-        registered_function_job_patch=RegisteredFunctionJobPatch(
-            status=RunningState.RUNNING,
-            task_id=task_uuid,
-        ),
+        function_class=FunctionClass.PROJECT,
+        job_creation_task_id=TaskID(task_uuid),
     )
 
 

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

@@ -107,6 +107,7 @@ async def run_function(
     )
 
     return await function_job_service.run_function(
+        job_creation_task_id=task_id,
         function=function,
         job_inputs=job_inputs,
         pricing_spec=pricing_spec,