Add function_id / jobs endpoint

Author: wvangeit

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/webserver/functions/functions_rpc_interface.py

@@ -124,13 +124,15 @@ async def list_function_jobs(
     *,
     pagination_limit: int,
     pagination_offset: int,
+    filter_by_function_id: FunctionID | None = None,
 ) -> tuple[list[RegisteredFunctionJob], PageMetaInfoLimitOffset]:
     result: tuple[list[RegisteredFunctionJob], PageMetaInfoLimitOffset] = (
         await rabbitmq_rpc_client.request(
             WEBSERVER_RPC_NAMESPACE,
             TypeAdapter(RPCMethodName).validate_python("list_function_jobs"),
             pagination_offset=pagination_offset,
             pagination_limit=pagination_limit,
+            filter_by_function_id=filter_by_function_id,
         )
     )
     return TypeAdapter(

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

@@ -99,6 +99,29 @@ async def list_functions(
     )
 
 
+@function_router.get(
+    "/{function_id:uuid}/jobs",
+    response_model=Page[RegisteredFunctionJob],
+    description="List function jobs for a function",
+)
+async def list_function_jobs(
+    function_id: FunctionID,
+    wb_api_rpc: Annotated[WbApiRpcClient, Depends(get_wb_api_rpc_client)],
+    page_params: Annotated[PaginationParams, Depends()],
+):
+    function_jobs_list, meta = await wb_api_rpc.list_function_jobs(
+        pagination_offset=page_params.offset,
+        pagination_limit=page_params.limit,
+        filter_by_function_id=function_id,
+    )
+
+    return create_page(
+        function_jobs_list,
+        total=meta.total,
+        params=page_params,
+    )
+
+
 @function_router.patch(
     "/{function_id:uuid}/title",
     response_model=RegisteredFunction,

services/api-server/src/simcore_service_api_server/services_rpc/wb_api_server.py

@@ -310,11 +310,13 @@ async def list_function_jobs(
         *,
         pagination_offset: PageOffsetInt = 0,
         pagination_limit: PageLimitInt = DEFAULT_NUMBER_OF_ITEMS_PER_PAGE,
+        filter_by_function_id: FunctionID | None = None,
     ) -> tuple[list[RegisteredFunctionJob], PageMetaInfoLimitOffset]:
         return await functions_rpc_interface.list_function_jobs(
             self._client,
             pagination_offset=pagination_offset,
             pagination_limit=pagination_limit,
+            filter_by_function_id=filter_by_function_id,
         )
 
     async def list_function_job_collections(

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

@@ -307,6 +307,35 @@ async def test_list_function_jobs(
     )
 
 
+async def test_list_function_jobs_with_function_filter(
+    client: AsyncClient,
+    mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],
+    sample_registered_function_job: RegisteredProjectFunctionJob,
+    sample_registered_function: RegisteredProjectFunction,
+) -> None:
+
+    mock_handler_in_functions_rpc_interface(
+        "list_function_jobs",
+        (
+            [sample_registered_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/{sample_registered_function.uid}/jobs"
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    data = response.json()["items"]
+    assert len(data) == 5
+    assert (
+        RegisteredProjectFunctionJob.model_validate(data[0])
+        == sample_registered_function_job
+    )
+
+
 async def test_delete_function_job(
     client: AsyncClient,
     mock_handler_in_functions_rpc_interface: Callable[[str, Any], None],

services/web/server/src/simcore_service_webserver/functions/_controller/_functions_rpc.py

@@ -100,11 +100,13 @@ async def list_function_jobs(
     app: web.Application,
     pagination_limit: int,
     pagination_offset: int,
+    filter_by_function_id: FunctionID | None = None,
 ) -> tuple[list[RegisteredFunctionJob], PageMetaInfoLimitOffset]:
     return await _functions_service.list_function_jobs(
         app=app,
         pagination_limit=pagination_limit,
         pagination_offset=pagination_offset,
+        filter_by_function_id=filter_by_function_id,
     )
 
 

services/web/server/src/simcore_service_webserver/functions/_functions_repository.py

@@ -229,16 +229,33 @@ async def list_function_jobs(
     *,
     pagination_limit: int,
     pagination_offset: int,
+    filter_by_function_id: FunctionID | None = None,
 ) -> tuple[list[RegisteredFunctionJobDB], PageMetaInfoLimitOffset]:
 
     async with transaction_context(get_asyncpg_engine(app), connection) as conn:
         total_count_result = await conn.scalar(
-            func.count().select().select_from(function_jobs_table)
+            func.count()
+            .select()
+            .select_from(function_jobs_table)
+            .where(
+                (
+                    function_jobs_table.c.function_uuid == filter_by_function_id
+                    if filter_by_function_id
+                    else True
+                ),
+            )
         )
         result = await conn.stream(
             function_jobs_table.select()
             .offset(pagination_offset)
             .limit(pagination_limit)
+            .where(
+                (
+                    function_jobs_table.c.function_uuid == filter_by_function_id
+                    if filter_by_function_id
+                    else True
+                ),
+            )
         )
         rows = await result.all()
         if rows is None:

services/web/server/src/simcore_service_webserver/functions/_functions_service.py

@@ -159,11 +159,13 @@ async def list_function_jobs(
     app: web.Application,
     pagination_limit: int,
     pagination_offset: int,
+    filter_by_function_id: FunctionID | None = None,
 ) -> tuple[list[RegisteredFunctionJob], PageMetaInfoLimitOffset]:
     returned_function_jobs, page = await _functions_repository.list_function_jobs(
         app=app,
         pagination_limit=pagination_limit,
         pagination_offset=pagination_offset,
+        filter_by_function_id=filter_by_function_id,
     )
     return [
         _decode_functionjob(returned_function_job)

services/web/server/tests/unit/with_dbs/04/functions_rpc/test_functions_controller_rpc.py

@@ -449,6 +449,65 @@ async def test_list_function_jobs(
     assert any(j.uid == registered_job.uid for j in jobs)
 
 
+async def test_list_function_jobs_for_functionid(
+    client: TestClient, rpc_client: RabbitMQRPCClient, mock_function: ProjectFunction
+):
+    # Register the function first
+    first_registered_function = await functions_rpc.register_function(
+        rabbitmq_rpc_client=rpc_client, function=mock_function
+    )
+    second_registered_function = await functions_rpc.register_function(
+        rabbitmq_rpc_client=rpc_client, function=mock_function
+    )
+
+    first_registered_function_jobs = []
+    second_registered_function_jobs = []
+    for i_job in range(6):
+        if i_job < 3:
+            function_job = ProjectFunctionJob(
+                function_uid=first_registered_function.uid,
+                title="Test Function Job",
+                description="A test function job",
+                project_job_id=uuid4(),
+                inputs={"input1": "value1"},
+                outputs={"output1": "result1"},
+            )
+            # Register the function job
+            first_registered_function_jobs.append(
+                await functions_rpc.register_function_job(
+                    rabbitmq_rpc_client=rpc_client, function_job=function_job
+                )
+            )
+        else:
+            function_job = ProjectFunctionJob(
+                function_uid=second_registered_function.uid,
+                title="Test Function Job",
+                description="A test function job",
+                project_job_id=uuid4(),
+                inputs={"input1": "value1"},
+                outputs={"output1": "result1"},
+            )
+            # Register the function job
+            second_registered_function_jobs.append(
+                await functions_rpc.register_function_job(
+                    rabbitmq_rpc_client=rpc_client, function_job=function_job
+                )
+            )
+
+    # List function jobs for a specific function ID
+    jobs, _ = await functions_rpc.list_function_jobs(
+        rabbitmq_rpc_client=rpc_client,
+        pagination_limit=10,
+        pagination_offset=0,
+        filter_by_function_id=first_registered_function.uid,
+    )
+
+    # Assert the list contains the registered job
+    assert len(jobs) > 0
+    assert len(jobs) == 3
+    assert all(j.function_uid == first_registered_function.uid for j in jobs)
+
+
 async def test_delete_function_job(
     client: TestClient, mock_function: ProjectFunction, rpc_client: RabbitMQRPCClient
 ):