Fix some pylint errors

Author: wvangeit

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

@@ -56,6 +56,26 @@ async def get_function_from_functionjob(
     )
 
 
+async def get_function_from_functionjobid(
+    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)],
+) -> RegisteredFunction:
+    function_job = await get_function_job_dependency(
+        function_job_id=function_job_id,
+        wb_api_rpc=wb_api_rpc,
+        user_id=user_id,
+        product_name=product_name,
+    )
+    return await get_function_from_functionjob(
+        function_job=function_job,
+        wb_api_rpc=wb_api_rpc,
+        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)],

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

@@ -12,7 +12,13 @@
     RegisteredFunctionJobCollection,
 )
 from models_library.products import ProductName
-from models_library.users import UserID  # Import UserID
+from models_library.users import UserID
+from simcore_service_api_server.api.dependencies.functions import (
+    get_stored_job_status,  # Import UserID
+)
+from simcore_service_api_server.api.dependencies.functions import (
+    get_function_from_functionjobid,
+)
 
 from ...models.pagination import Page, PaginationParams
 from ...models.schemas.errors import ErrorGet
@@ -221,13 +227,30 @@ async def function_job_collection_status(
     job_statuses = await asyncio.gather(
         *[
             function_job_status(
-                job_id,
+                function_job=await get_function_job(
+                    function_job_id=function_job_id,
+                    wb_api_rpc=wb_api_rpc,
+                    user_id=user_id,
+                    product_name=product_name,
+                ),
+                function=await get_function_from_functionjobid(
+                    function_job_id=function_job_id,
+                    wb_api_rpc=wb_api_rpc,
+                    user_id=user_id,
+                    product_name=product_name,
+                ),
+                stored_job_status=await get_stored_job_status(
+                    function_job_id=function_job_id,
+                    user_id=user_id,
+                    product_name=product_name,
+                    wb_api_rpc=wb_api_rpc,
+                ),
                 wb_api_rpc=wb_api_rpc,
                 director2_api=director2_api,
                 user_id=user_id,
                 product_name=product_name,
             )
-            for job_id in function_job_collection.job_ids
+            for function_job_id in function_job_collection.job_ids
         ]
     )
     return FunctionJobCollectionStatus(

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

@@ -309,301 +309,6 @@ async def test_delete_function(
     assert response.status_code == status.HTTP_200_OK
 
 
-async def test_register_function_job(
-    client: AsyncClient,
-    mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],
-    mock_project_function_job: ProjectFunctionJob,
-    mock_registered_project_function_job: RegisteredProjectFunctionJob,
-    auth: httpx.BasicAuth,
-) -> None:
-    """Test the register_function_job endpoint."""
-
-    mock_handler_in_functions_rpc_interface(
-        "register_function_job", mock_registered_project_function_job
-    )
-
-    response = await client.post(
-        f"{API_VTAG}/function_jobs",
-        json=mock_project_function_job.model_dump(mode="json"),
-        auth=auth,
-    )
-
-    assert response.status_code == status.HTTP_200_OK
-    assert (
-        RegisteredProjectFunctionJob.model_validate(response.json())
-        == mock_registered_project_function_job
-    )
-
-
-async def test_get_function_job(
-    client: AsyncClient,
-    mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],
-    mock_registered_project_function_job: RegisteredProjectFunctionJob,
-    auth: httpx.BasicAuth,
-) -> None:
-    mock_handler_in_functions_rpc_interface(
-        "get_function_job", mock_registered_project_function_job
-    )
-
-    # Now, get the function job
-    response = await client.get(
-        f"{API_VTAG}/function_jobs/{mock_registered_project_function_job.uid}",
-        auth=auth,
-    )
-    assert response.status_code == status.HTTP_200_OK
-    assert (
-        RegisteredProjectFunctionJob.model_validate(response.json())
-        == mock_registered_project_function_job
-    )
-
-
-async def test_list_function_jobs(
-    client: AsyncClient,
-    mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],
-    mock_registered_project_function_job: RegisteredProjectFunctionJob,
-    auth: httpx.BasicAuth,
-) -> None:
-    mock_handler_in_functions_rpc_interface(
-        "list_function_jobs",
-        (
-            [mock_registered_project_function_job for _ in range(5)],
-            PageMetaInfoLimitOffset(total=5, count=5, limit=10, offset=0),
-        ),
-    )
-
-    # Now, list function jobs
-    response = await client.get(f"{API_VTAG}/function_jobs", auth=auth)
-    assert response.status_code == status.HTTP_200_OK
-    data = response.json()["items"]
-    assert len(data) == 5
-    assert (
-        RegisteredProjectFunctionJob.model_validate(data[0])
-        == mock_registered_project_function_job
-    )
-
-
-async def test_list_function_jobs_with_function_filter(
-    client: AsyncClient,
-    mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],
-    mock_registered_project_function_job: RegisteredProjectFunctionJob,
-    mock_registered_project_function: RegisteredProjectFunction,
-    auth: httpx.BasicAuth,
-) -> None:
-    mock_handler_in_functions_rpc_interface(
-        "list_function_jobs",
-        (
-            [mock_registered_project_function_job for _ in range(5)],
-            PageMetaInfoLimitOffset(total=5, count=5, limit=10, offset=0),
-        ),
-    )
-
-    # Now, list function jobs with a filter
-    response = await client.get(
-        f"{API_VTAG}/functions/{mock_registered_project_function.uid}/jobs", auth=auth
-    )
-
-    assert response.status_code == status.HTTP_200_OK
-    data = response.json()["items"]
-    assert len(data) == 5
-    assert (
-        RegisteredProjectFunctionJob.model_validate(data[0])
-        == mock_registered_project_function_job
-    )
-
-
-async def test_delete_function_job(
-    client: AsyncClient,
-    mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],
-    mock_registered_project_function_job: RegisteredProjectFunctionJob,
-    auth: httpx.BasicAuth,
-) -> None:
-    mock_handler_in_functions_rpc_interface("delete_function_job", None)
-
-    # Now, delete the function job
-    response = await client.delete(
-        f"{API_VTAG}/function_jobs/{mock_registered_project_function_job.uid}",
-        auth=auth,
-    )
-    assert response.status_code == status.HTTP_200_OK
-
-
-async def test_register_function_job_collection(
-    client: AsyncClient,
-    mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],
-    auth: httpx.BasicAuth,
-) -> None:
-    mock_function_job_collection = FunctionJobCollection.model_validate(
-        {
-            "title": "Test Collection",
-            "description": "A test function job collection",
-            "job_ids": [str(uuid4()), str(uuid4())],
-        }
-    )
-
-    mock_registered_function_job_collection = (
-        RegisteredFunctionJobCollection.model_validate(
-            {
-                **mock_function_job_collection.model_dump(),
-                "uid": str(uuid4()),
-                "created_at": datetime.datetime.now(datetime.UTC),
-            }
-        )
-    )
-
-    mock_handler_in_functions_rpc_interface(
-        "register_function_job_collection", mock_registered_function_job_collection
-    )
-
-    response = await client.post(
-        f"{API_VTAG}/function_job_collections",
-        json=mock_function_job_collection.model_dump(mode="json"),
-        auth=auth,
-    )
-
-    # Assert
-    assert response.status_code == status.HTTP_200_OK
-    assert (
-        RegisteredFunctionJobCollection.model_validate(response.json())
-        == mock_registered_function_job_collection
-    )
-
-
-async def test_get_function_job_collection(
-    client: AsyncClient,
-    mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],
-    auth: httpx.BasicAuth,
-) -> None:
-    mock_registered_function_job_collection = (
-        RegisteredFunctionJobCollection.model_validate(
-            {
-                "uid": str(uuid4()),
-                "title": "Test Collection",
-                "description": "A test function job collection",
-                "job_ids": [str(uuid4()), str(uuid4())],
-                "created_at": datetime.datetime.now(datetime.UTC),
-            }
-        )
-    )
-
-    mock_handler_in_functions_rpc_interface(
-        "get_function_job_collection", mock_registered_function_job_collection
-    )
-
-    response = await client.get(
-        f"{API_VTAG}/function_job_collections/{mock_registered_function_job_collection.uid}",
-        auth=auth,
-    )
-    assert response.status_code == status.HTTP_200_OK
-    assert (
-        RegisteredFunctionJobCollection.model_validate(response.json())
-        == mock_registered_function_job_collection
-    )
-
-
-async def test_list_function_job_collections(
-    client: AsyncClient,
-    mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],
-    auth: httpx.BasicAuth,
-) -> None:
-    mock_registered_function_job_collection = (
-        RegisteredFunctionJobCollection.model_validate(
-            {
-                "uid": str(uuid4()),
-                "title": "Test Collection",
-                "description": "A test function job collection",
-                "job_ids": [str(uuid4()), str(uuid4())],
-                "created_at": datetime.datetime.now(datetime.UTC),
-            }
-        )
-    )
-
-    mock_handler_in_functions_rpc_interface(
-        "list_function_job_collections",
-        (
-            [mock_registered_function_job_collection for _ in range(5)],
-            PageMetaInfoLimitOffset(total=5, count=5, limit=10, offset=0),
-        ),
-    )
-
-    response = await client.get(f"{API_VTAG}/function_job_collections", auth=auth)
-    assert response.status_code == status.HTTP_200_OK
-    data = response.json()["items"]
-    assert len(data) == 5
-    assert (
-        RegisteredFunctionJobCollection.model_validate(data[0])
-        == mock_registered_function_job_collection
-    )
-
-
-async def test_delete_function_job_collection(
-    client: AsyncClient,
-    mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],
-    mock_registered_function_job_collection: RegisteredFunctionJobCollection,
-    auth: httpx.BasicAuth,
-) -> None:
-    mock_handler_in_functions_rpc_interface("delete_function_job_collection", None)
-
-    # Now, delete the function job collection
-    response = await client.delete(
-        f"{API_VTAG}/function_job_collections/{mock_registered_function_job_collection.uid}",
-        auth=auth,
-    )
-    assert response.status_code == status.HTTP_200_OK
-    data = response.json()
-    assert data is None
-
-
-async def test_get_function_job_collection_jobs(
-    client: AsyncClient,
-    mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],
-    mock_registered_function_job_collection: RegisteredFunctionJobCollection,
-    auth: httpx.BasicAuth,
-) -> None:
-    mock_handler_in_functions_rpc_interface(
-        "get_function_job_collection", mock_registered_function_job_collection
-    )
-
-    response = await client.get(
-        f"{API_VTAG}/function_job_collections/{mock_registered_function_job_collection.uid}/function_jobs",
-        auth=auth,
-    )
-    assert response.status_code == status.HTTP_200_OK
-    data = response.json()
-    assert len(data) == len(mock_registered_function_job_collection.job_ids)
-
-
-async def test_list_function_job_collections_with_function_filter(
-    client: AsyncClient,
-    mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],
-    mock_registered_function_job_collection: RegisteredFunctionJobCollection,
-    mock_registered_project_function: RegisteredProjectFunction,
-    auth: httpx.BasicAuth,
-) -> None:
-    mock_handler_in_functions_rpc_interface(
-        "list_function_job_collections",
-        (
-            [mock_registered_function_job_collection for _ in range(2)],
-            PageMetaInfoLimitOffset(total=5, count=2, limit=2, offset=1),
-        ),
-    )
-
-    response = await client.get(
-        f"{API_VTAG}/function_job_collections?function_id={mock_registered_project_function.uid}&limit=2&offset=1",
-        auth=auth,
-    )
-    assert response.status_code == status.HTTP_200_OK
-    data = response.json()
-
-    assert data["total"] == 5
-    assert data["limit"] == 2
-    assert data["offset"] == 1
-    assert len(data["items"]) == 2
-    assert (
-        RegisteredFunctionJobCollection.model_validate(data["items"][0])
-        == mock_registered_function_job_collection
-    )
-
-
 @pytest.mark.parametrize("user_has_execute_right", [False, True])
 @pytest.mark.parametrize(
     "funcapi_endpoint,endpoint_inputs", [("run", {}), ("map", [{}, {}])]