Move some code in dependencies

Author: wvangeit

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

@@ -0,0 +1,67 @@
+from typing import Annotated
+
+from fastapi import Depends
+from models_library.functions import (
+    FunctionJob,
+    FunctionJobID,
+    FunctionJobStatus,
+    FunctionOutputs,
+    RegisteredFunction,
+)
+from models_library.products import ProductName
+from models_library.users import UserID
+from simcore_service_api_server.api.dependencies.authentication import (
+    get_current_user_id,
+    get_product_name,
+)
+from simcore_service_api_server.api.dependencies.webserver_rpc import (
+    get_wb_api_rpc_client,
+)
+from simcore_service_api_server.services_rpc.wb_api_server import WbApiRpcClient
+
+
+async def get_stored_job_outputs(
+    function_job_id: FunctionJobID,
+    wb_api_rpc: 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)],
+) -> FunctionOutputs:
+
+    return await wb_api_rpc.get_function_job_outputs(
+        function_job_id=function_job_id, user_id=user_id, product_name=product_name
+    )
+
+
+async def get_function_job_dependency(
+    function_job_id: FunctionJobID,
+    wb_api_rpc: 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)],
+) -> FunctionJob:
+    return await wb_api_rpc.get_function_job(
+        function_job_id=function_job_id, user_id=user_id, product_name=product_name
+    )
+
+
+async def get_function_from_functionjob(
+    function_job: Annotated[FunctionJob, Depends(get_function_job_dependency)],
+    wb_api_rpc: 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)],
+) -> RegisteredFunction:
+    return await wb_api_rpc.get_function(
+        function_id=function_job.function_uid,
+        user_id=user_id,
+        product_name=product_name,
+    )
+
+
+async def get_stored_job_status(
+    function_job_id: FunctionJobID,
+    wb_api_rpc: 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)],
+) -> FunctionJobStatus:
+    return await wb_api_rpc.get_function_job_status(
+        function_job_id=function_job_id, user_id=user_id, product_name=product_name
+    )

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

@@ -20,6 +20,12 @@
 from models_library.projects_state import RunningState
 from models_library.users import UserID
 from servicelib.fastapi.dependencies import get_app
+from simcore_service_api_server.api.dependencies.functions import (
+    get_function_from_functionjob,
+    get_function_job_dependency,
+    get_stored_job_outputs,
+    get_stored_job_status,
+)
 from sqlalchemy.ext.asyncio import AsyncEngine
 
 from ..._service_jobs import JobService
@@ -172,25 +178,19 @@ async def delete_function_job(
     ),
 )
 async def function_job_status(
-    function_job_id: FunctionJobID,
+    function_job: Annotated[
+        RegisteredFunctionJob, Depends(get_function_job_dependency)
+    ],
+    function: Annotated[RegisteredFunction, Depends(get_function_from_functionjob)],
+    stored_job_status: Annotated[FunctionJobStatus, Depends(get_stored_job_status)],
     director2_api: Annotated[DirectorV2Api, Depends(get_api_client(DirectorV2Api))],
     user_id: Annotated[UserID, Depends(get_current_user_id)],
     product_name: Annotated[ProductName, Depends(get_product_name)],
     wb_api_rpc: Annotated[WbApiRpcClient, Depends(get_wb_api_rpc_client)],
 ) -> FunctionJobStatus:
 
-    function, function_job = await get_function_from_functionjobid(
-        wb_api_rpc=wb_api_rpc,
-        function_job_id=function_job_id,
-        user_id=user_id,
-        product_name=product_name,
-    )
-    old_job_status = await wb_api_rpc.get_function_job_status(
-        function_job_id=function_job.uid, user_id=user_id, product_name=product_name
-    )
-
-    if old_job_status.status in (RunningState.SUCCESS, RunningState.FAILED):
-        return old_job_status
+    if stored_job_status.status in (RunningState.SUCCESS, RunningState.FAILED):
+        return stored_job_status
 
     if (
         function.function_class == FunctionClass.PROJECT
@@ -264,27 +264,20 @@ async def get_function_from_functionjobid(
     ),
 )
 async def get_function_job_outputs(
-    function_job_id: FunctionJobID,
+    function_job: Annotated[
+        RegisteredFunctionJob, Depends(get_function_job_dependency)
+    ],
+    function: Annotated[RegisteredFunction, Depends(get_function_from_functionjob)],
     webserver_api: Annotated[AuthSession, Depends(get_webserver_session)],
     user_id: Annotated[UserID, Depends(get_current_user_id)],
     product_name: Annotated[ProductName, Depends(get_product_name)],
     storage_client: Annotated[StorageApi, Depends(get_api_client(StorageApi))],
     wb_api_rpc: Annotated[WbApiRpcClient, Depends(get_wb_api_rpc_client)],
     async_pg_engine: Annotated[AsyncEngine, Depends(get_db_asyncpg_engine)],
+    stored_job_outputs: Annotated[FunctionOutputs, Depends(get_stored_job_outputs)],
 ) -> FunctionOutputs:
-    function, function_job = await get_function_from_functionjobid(
-        wb_api_rpc=wb_api_rpc,
-        function_job_id=function_job_id,
-        user_id=user_id,
-        product_name=product_name,
-    )
-
-    old_job_outputs = await wb_api_rpc.get_function_job_outputs(
-        function_job_id=function_job.uid, user_id=user_id, product_name=product_name
-    )
-
-    if old_job_outputs is not None:
-        return old_job_outputs
+    if stored_job_outputs is not None:
+        return stored_job_outputs
 
     if (
         function.function_class == FunctionClass.PROJECT

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

@@ -36,6 +36,7 @@
 from models_library.users import UserID
 from servicelib.fastapi.dependencies import get_reverse_url_mapper
 from simcore_service_api_server._service_jobs import JobService
+from simcore_service_api_server.api.dependencies.functions import get_stored_job_status
 
 from ..._service_solvers import SolverService
 from ...models.pagination import Page, PaginationParams
@@ -365,11 +366,12 @@ async def validate_function_inputs(
 )
 async def run_function(  # noqa: PLR0913
     request: Request,
+    function_id: FunctionID,
+    to_run_function: Annotated[RegisteredFunction, Depends(get_function)],
     wb_api_rpc: Annotated[WbApiRpcClient, Depends(get_wb_api_rpc_client)],
     webserver_api: Annotated[AuthSession, Depends(get_webserver_session)],
     url_for: Annotated[Callable, Depends(get_reverse_url_mapper)],
     director2_api: Annotated[DirectorV2Api, Depends(get_api_client(DirectorV2Api))],
-    function_id: FunctionID,
     function_inputs: FunctionInputs,
     user_id: Annotated[UserID, Depends(get_current_user_id)],
     product_name: Annotated[str, Depends(get_product_name)],
@@ -378,7 +380,6 @@ 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:
-
     parent_project_uuid = (
         x_simcore_parent_project_uuid
         if isinstance(x_simcore_parent_project_uuid, ProjectID)
@@ -390,8 +391,6 @@ async def run_function(  # noqa: PLR0913
         else None
     )
 
-    # Make sure the user is allowed to execute any function
-    # (read/write right is checked in the other endpoint called in this method)
     user_api_access_rights = await wb_api_rpc.get_functions_user_api_access_rights(
         user_id=user_id, product_name=product_name
     )
@@ -400,8 +399,7 @@ async def run_function(  # noqa: PLR0913
             user_id=user_id,
             function_id=function_id,
         )
-    # Make sure the user is allowed to execute this particular function
-    # (read/write right is checked in the other endpoint called in this method)
+
     user_permissions: FunctionUserAccessRights = (
         await wb_api_rpc.get_function_user_permissions(
             function_id=function_id, user_id=user_id, product_name=product_name
@@ -415,10 +413,6 @@ async def run_function(  # noqa: PLR0913
 
     from .function_jobs_routes import function_job_status
 
-    to_run_function = await wb_api_rpc.get_function(
-        function_id=function_id, user_id=user_id, product_name=product_name
-    )
-
     joined_inputs = _join_inputs(
         to_run_function.default_inputs,
         function_inputs,
@@ -443,10 +437,17 @@ async def run_function(  # noqa: PLR0913
     ):
         for cached_function_job in cached_function_jobs:
             job_status = await function_job_status(
+                function=to_run_function,
+                function_job=cached_function_job,
+                stored_job_status=await get_stored_job_status(
+                    function_job_id=cached_function_job.uid,
+                    user_id=user_id,
+                    product_name=product_name,
+                    wb_api_rpc=wb_api_rpc,
+                ),
                 wb_api_rpc=wb_api_rpc,
-                director2_api=director2_api,
-                function_job_id=cached_function_job.uid,
                 user_id=user_id,
+                director2_api=director2_api,
                 product_name=product_name,
             )
             if job_status.status == RunningState.SUCCESS:
@@ -563,9 +564,10 @@ async def delete_function(
     ),
 )
 async def map_function(  # noqa: PLR0913
+    request: Request,
     function_id: FunctionID,
+    to_run_function: Annotated[RegisteredFunction, Depends(get_function)],
     function_inputs_list: FunctionInputsList,
-    request: Request,
     wb_api_rpc: Annotated[WbApiRpcClient, Depends(get_wb_api_rpc_client)],
     webserver_api: Annotated[AuthSession, Depends(get_webserver_session)],
     url_for: Annotated[Callable, Depends(get_reverse_url_mapper)],
@@ -577,11 +579,11 @@ async def map_function(  # noqa: PLR0913
     x_simcore_parent_project_uuid: Annotated[ProjectID | Literal["null"], Header()],
     x_simcore_parent_node_id: Annotated[NodeID | Literal["null"], Header()],
 ) -> RegisteredFunctionJobCollection:
-    function_jobs = []
     function_jobs = [
         await run_function(
             wb_api_rpc=wb_api_rpc,
             function_id=function_id,
+            to_run_function=to_run_function,
             function_inputs=function_inputs,
             product_name=product_name,
             user_id=user_id,

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

@@ -142,19 +142,17 @@ def mock_registered_solver_function(
     sample_output_schema: JSONFunctionOutputSchema,
 ) -> RegisteredFunction:
     return RegisteredSolverFunction(
-        **{
-            "title": "test_function",
-            "function_class": FunctionClass.SOLVER,
-            "description": "A test function",
-            "input_schema": sample_input_schema,
-            "output_schema": sample_output_schema,
-            "default_inputs": None,
-            "uid": f"{uuid4()}",
-            "created_at": datetime.datetime.now(datetime.UTC),
-            "modified_at": datetime.datetime.now(datetime.UTC),
-            "solver_key": "simcore/services/comp/ans-model",
-            "solver_version": "1.0.1",
-        }
+        title="test_function",
+        function_class=FunctionClass.SOLVER,
+        description="A test function",
+        input_schema=sample_input_schema,
+        output_schema=sample_output_schema,
+        default_inputs=None,
+        uid=uuid4(),
+        created_at=datetime.datetime.now(datetime.UTC),
+        modified_at=datetime.datetime.now(datetime.UTC),
+        solver_key="simcore/services/comp/ans-model",
+        solver_version="1.0.1",
     )