factor out function run pre check

Author: bisgaard-itis

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

@@ -160,17 +160,12 @@ async def inspect_function_job(
             job_status=new_job_status,
         )
 
-    async def run_function(
+    async def run_function_pre_check(
         self,
         *,
         function: RegisteredFunction,
         function_inputs: FunctionInputs,
-        pricing_spec: JobPricingSpecification | None,
-        job_links: JobLinks,
-        x_simcore_parent_project_uuid: NodeID | None,
-        x_simcore_parent_node_id: NodeID | None,
-    ) -> RegisteredFunctionJob:
-
+    ) -> JobInputs:
         user_api_access_rights = (
             await self._web_rpc_client.get_functions_user_api_access_rights(
                 user_id=self.user_id, product_name=self.product_name
@@ -206,9 +201,24 @@ async def run_function(
             if not is_valid:
                 raise FunctionInputsValidationError(error=validation_str)
 
+        return JobInputs(
+            values=joined_inputs or {},
+        )
+
+    async def run_function(
+        self,
+        *,
+        function: RegisteredFunction,
+        job_inputs: JobInputs,
+        pricing_spec: JobPricingSpecification | None,
+        job_links: JobLinks,
+        x_simcore_parent_project_uuid: NodeID | None,
+        x_simcore_parent_node_id: NodeID | None,
+    ) -> RegisteredFunctionJob:
+
         if cached_function_jobs := await self._web_rpc_client.find_cached_function_jobs(
             function_id=function.uid,
-            inputs=joined_inputs,
+            inputs=job_inputs.values,
             user_id=self.user_id,
             product_name=self.product_name,
         ):
@@ -223,7 +233,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=JobInputs(values=joined_inputs or {}),
+                job_inputs=job_inputs,
                 hidden=True,
                 job_links=job_links,
                 x_simcore_parent_project_uuid=x_simcore_parent_project_uuid,
@@ -239,7 +249,7 @@ async def run_function(
                     function_uid=function.uid,
                     title=f"Function job of function {function.uid}",
                     description=function.description,
-                    inputs=joined_inputs,
+                    inputs=job_inputs.values,
                     outputs=None,
                     project_job_id=study_job.id,
                 ),
@@ -251,7 +261,7 @@ async def run_function(
             solver_job = await self._job_service.create_solver_job(
                 solver_key=function.solver_key,
                 version=function.solver_version,
-                inputs=JobInputs(values=joined_inputs or {}),
+                inputs=job_inputs,
                 job_links=job_links,
                 hidden=True,
                 x_simcore_parent_project_uuid=x_simcore_parent_project_uuid,
@@ -268,7 +278,7 @@ async def run_function(
                     function_uid=function.uid,
                     title=f"Function job of function {function.uid}",
                     description=function.description,
-                    inputs=joined_inputs,
+                    inputs=job_inputs.values,
                     outputs=None,
                     solver_job_id=solver_job.id,
                 ),
@@ -291,16 +301,24 @@ async def map_function(
         x_simcore_parent_node_id: NodeID | None,
     ) -> RegisteredFunctionJobCollection:
 
+        job_inputs = [
+            await self.run_function_pre_check(
+                function=function,
+                function_inputs=inputs,
+            )
+            for inputs in function_inputs_list
+        ]
+
         function_jobs = [
             await self.run_function(
                 function=function,
-                function_inputs=function_inputs,
+                job_inputs=inputs,
                 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 function_inputs in function_inputs_list
+            for inputs in job_inputs
         ]
 
         function_job_collection_description = f"Function job collection of map of function {function.uid} with {len(function_inputs_list)} inputs"

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

@@ -328,6 +328,9 @@ async def run_function(  # noqa: PLR0913
     url_for: Annotated[Callable, Depends(get_reverse_url_mapper)],
     function_inputs: FunctionInputs,
     function_service: Annotated[FunctionService, Depends(get_function_service)],
+    function_job_service: Annotated[
+        FunctionJobService, Depends(get_function_job_service)
+    ],
     x_simcore_parent_project_uuid: Annotated[ProjectID | Literal["null"], Header()],
     x_simcore_parent_node_id: Annotated[NodeID | Literal["null"], Header()],
 ) -> TaskGet:
@@ -345,6 +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.run_function_pre_check(
+        function=to_run_function, function_inputs=function_inputs
+    )
+
     job_filter = AsyncJobFilter(
         user_id=user_identity.user_id,
         product_name=user_identity.product_name,
@@ -362,7 +369,7 @@ async def run_function(  # noqa: PLR0913
         task_filter=task_filter,
         user_identity=user_identity,
         function=to_run_function,
-        function_inputs=function_inputs,
+        job_inputs=job_inputs,
         pricing_spec=pricing_spec,
         job_links=job_links,
         x_simcore_parent_project_uuid=parent_project_uuid,

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

@@ -3,9 +3,10 @@
 )
 from celery_library.utils import get_app_server  # pylint: disable=no-name-in-module
 from fastapi import FastAPI
-from models_library.functions import FunctionInputs, 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
 
 from ...api.dependencies.authentication import Identity
 from ...api.dependencies.rabbitmq import get_rabbitmq_rpc_client
@@ -20,12 +21,14 @@
 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 JobPricingSpecification
+from ...models.schemas.jobs import JobInputs, JobPricingSpecification
 from ...services_http.director_v2 import DirectorV2Api
 from ...services_http.storage import StorageApi
 
 
-async def _assemble_function_job_service(*, app: FastAPI, user_identity: Identity):
+async def _assemble_function_job_service(
+    *, app: FastAPI, user_identity: Identity
+) -> FunctionJobService:
     # to avoid this show we could introduce a dependency injection
     # system which is not linked to FastAPI (i.e. can be resolved manually).
     # One suggestion: https://github.com/ets-labs/python-dependency-injector, which is compatible
@@ -91,7 +94,7 @@ async def run_function(
     *,
     user_identity: Identity,
     function: RegisteredFunction,
-    function_inputs: FunctionInputs,
+    job_inputs: JobInputs,
     pricing_spec: JobPricingSpecification | None,
     job_links: JobLinks,
     x_simcore_parent_project_uuid: NodeID | None,
@@ -105,7 +108,7 @@ async def run_function(
 
     return await function_job_service.run_function(
         function=function,
-        function_inputs=function_inputs,
+        job_inputs=job_inputs,
         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/celery/worker_tasks/tasks.py

@@ -16,13 +16,14 @@
 
 from ...api.dependencies.authentication import Identity
 from ...models.api_resources import JobLinks
-from ...models.schemas.jobs import JobPricingSpecification
+from ...models.schemas.jobs import JobInputs, JobPricingSpecification
 from .functions_tasks import run_function
 
 _logger = logging.getLogger(__name__)
 
 pydantic_types_to_register = (
     Identity,
+    JobInputs,
     JobLinks,
     JobPricingSpecification,
     RegisteredProjectFunction,

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

@@ -31,7 +31,6 @@
 from models_library.functions import (
     FunctionClass,
     FunctionID,
-    FunctionInputs,
     FunctionJobID,
     FunctionUserAccessRights,
     FunctionUserApiAccessRights,
@@ -42,26 +41,31 @@
 )
 from models_library.projects import ProjectID
 from models_library.users import UserID
-from pytest_mock import MockType
+from pytest_mock import MockerFixture, MockType
 from pytest_simcore.helpers.httpx_calls_capture_models import HttpApiCallCaptureModel
 from servicelib.celery.models import TaskFilter, TaskID, TaskMetadata, TasksQueue
 from servicelib.common_headers import (
     X_SIMCORE_PARENT_NODE_ID,
     X_SIMCORE_PARENT_PROJECT_UUID,
 )
 from simcore_service_api_server._meta import API_VTAG
+from simcore_service_api_server._service_function_jobs import FunctionJobService
 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.api.dependencies.services import (
+    get_function_job_service,
+)
 from simcore_service_api_server.api.routes.functions_routes import get_function
 from simcore_service_api_server.celery.worker_tasks.functions_tasks import (
     run_function as run_function_task,
 )
 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.schemas.jobs import (
+    JobInputs,
     JobPricingSpecification,
     NodeID,
 )
@@ -112,7 +116,7 @@ async def run_function(
         *,
         user_identity: Identity,
         function: RegisteredFunction,
-        function_inputs: FunctionInputs,
+        job_inputs: JobInputs,
         pricing_spec: JobPricingSpecification | None,
         job_links: JobLinks,
         x_simcore_parent_project_uuid: NodeID | None,
@@ -122,7 +126,7 @@ async def run_function(
             title=_faker.sentence(),
             description=_faker.paragraph(),
             function_uid=FunctionID(_faker.uuid4()),
-            inputs=function_inputs,
+            inputs=job_inputs.values,
             outputs=None,
             function_class=FunctionClass.PROJECT,
             uid=FunctionJobID(_faker.uuid4()),
@@ -149,19 +153,9 @@ async def test_with_fake_run_function(
     app: FastAPI,
     client: AsyncClient,
     auth: BasicAuth,
+    mocker: MockerFixture,
     with_api_server_celery_worker: TestWorkController,
 ):
-    app.dependency_overrides[get_function] = (
-        lambda: RegisteredProjectFunction.model_validate(
-            RegisteredProjectFunction.model_config.get("json_schema_extra", {}).get(
-                "examples", []
-            )[0]
-        )
-    )
-
-    headers = {}
-    headers[X_SIMCORE_PARENT_PROJECT_UUID] = "null"
-    headers[X_SIMCORE_PARENT_NODE_ID] = "null"
 
     body = {
         "input_1": _faker.uuid4(),
@@ -175,6 +169,25 @@ async def test_with_fake_run_function(
         ],
     }
 
+    async def mock_get_function_job_service() -> FunctionJobService:
+        mock = mocker.AsyncMock(spec=FunctionJobService)
+        mock.run_function_pre_check.return_value = JobInputs(values=body)
+        return mock
+
+    app.dependency_overrides[get_function_job_service] = mock_get_function_job_service
+
+    app.dependency_overrides[get_function] = (
+        lambda: RegisteredProjectFunction.model_validate(
+            RegisteredProjectFunction.model_config.get("json_schema_extra", {}).get(
+                "examples", []
+            )[0]
+        )
+    )
+
+    headers = {}
+    headers[X_SIMCORE_PARENT_PROJECT_UUID] = "null"
+    headers[X_SIMCORE_PARENT_NODE_ID] = "null"
+
     response = await client.post(
         f"/{API_VTAG}/functions/{_faker.uuid4()}:run",
         auth=auth,

services/api-server/tests/unit/api_functions/test_api_routers_functions.py

@@ -50,6 +50,7 @@
 from simcore_service_api_server.api.dependencies.authentication import Identity
 from simcore_service_api_server.celery.worker_tasks import functions_tasks
 from simcore_service_api_server.models.api_resources import JobLinks
+from simcore_service_api_server.models.schemas.jobs import JobInputs
 from simcore_service_api_server.services_rpc.wb_api_server import WbApiRpcClient
 
 _faker = Faker()
@@ -478,7 +479,7 @@ def _default_side_effect(
         task_id=TaskID(_faker.uuid4()),
         user_identity=user_identity,
         function=mock_registered_project_function,
-        function_inputs={},
+        job_inputs=JobInputs(values={}),
         pricing_spec=None,
         job_links=job_links,
         x_simcore_parent_project_uuid=None,