add filter filter_only_running - director-v2

Author: matusdrobuliak66

packages/postgres-database/src/simcore_postgres_database/migration/versions/42ec7816c0b4_computational_collection_runs.py

@@ -119,3 +119,4 @@ def downgrade():
     )
     op.drop_table("comp_runs_collections")
     # ### end Alembic commands ###
+    # ### end Alembic commands ###

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/director_v2/computations.py

@@ -126,6 +126,7 @@ async def list_computation_collection_runs_page(
     product_name: ProductName,
     user_id: UserID,
     project_ids: list[ProjectID] | None,
+    filter_only_running: bool = False,
     # pagination
     offset: int = 0,
     limit: int = 20,
@@ -138,6 +139,7 @@ async def list_computation_collection_runs_page(
         product_name=product_name,
         user_id=user_id,
         project_ids=project_ids,
+        filter_only_running=filter_only_running,
         offset=offset,
         limit=limit,
         timeout_s=_DEFAULT_TIMEOUT_S,

services/director-v2/src/simcore_service_director_v2/api/rpc/_computations.py

@@ -102,15 +102,24 @@ async def list_computation_collection_runs_page(
     product_name: ProductName,
     user_id: UserID,
     project_ids: list[ProjectID] | None,
+    filter_only_running: bool = False,
     # pagination
     offset: int = 0,
     limit: int = 20,
 ) -> ComputationCollectionRunRpcGetPage:
     comp_runs_repo = CompRunsRepository.instance(db_engine=app.state.engine)
+
+    collection_run_ids: list[CollectionRunID] | None = None
+    if filter_only_running is True:
+        collection_run_ids = await comp_runs_repo.list_all_collection_run_ids_for_user_currently_running_computations(
+            product_name=product_name, user_id=user_id
+        )
+
     total, comp_runs_output = await comp_runs_repo.list_group_by_collection_run_id(
         product_name=product_name,
         user_id=user_id,
-        project_ids=project_ids,
+        project_ids_or_none=project_ids,
+        collection_run_ids_or_none=collection_run_ids,
         offset=offset,
         limit=limit,
     )

services/director-v2/src/simcore_service_director_v2/modules/db/repositories/comp_runs.py

@@ -377,12 +377,46 @@ async def list_for_user_and_project_all_iterations(
 
             return cast(int, total_count), items
 
+    async def list_all_collection_run_ids_for_user_currently_running_computations(
+        self,
+        *,
+        product_name: str,
+        user_id: UserID,
+    ) -> list[CollectionRunID]:
+
+        list_query = (
+            sa.select(
+                comp_runs.c.collection_run_id,
+            )
+            .where(
+                (comp_runs.c.user_id == user_id)
+                & (
+                    comp_runs.c.metadata["product_name"].astext == product_name
+                )  # <-- NOTE: We might create a separate column for this for fast retrieval
+                & (
+                    comp_runs.c.result.in_(
+                        [
+                            RUNNING_STATE_TO_DB[item]
+                            for item in RunningState.list_running_states()
+                        ]
+                    )
+                )
+            )
+            .distinct()
+        )
+
+        async with pass_or_acquire_connection(self.db_engine) as conn:
+            return [
+                CollectionRunID(row[0]) async for row in await conn.stream(list_query)
+            ]
+
     async def list_group_by_collection_run_id(
         self,
         *,
         product_name: str,
         user_id: UserID,
-        project_ids: list[ProjectID] | None = None,
+        project_ids_or_none: list[ProjectID] | None = None,
+        collection_run_ids_or_none: list[CollectionRunID] | None = None,
         # pagination
         offset: int,
         limit: int,
@@ -408,10 +442,19 @@ async def list_group_by_collection_run_id(
             & (comp_runs.c.metadata["product_name"].astext == product_name)
         )
 
-        if project_ids:
+        if project_ids_or_none:
             base_select_query = base_select_query.where(
                 comp_runs.c.project_uuid.in_(
-                    [f"{project_id}" for project_id in project_ids]
+                    [f"{project_id}" for project_id in project_ids_or_none]
+                )
+            )
+        if collection_run_ids_or_none:
+            base_select_query = base_select_query.where(
+                comp_runs.c.collection_run_id.in_(
+                    [
+                        f"{collection_run_id}"
+                        for collection_run_id in collection_run_ids_or_none
+                    ]
                 )
             )
 

services/director-v2/tests/unit/with_dbs/comp_scheduler/test_api_rpc_computations.py

@@ -265,18 +265,29 @@ async def test_rpc_list_computation_collection_runs_page_and_collection_run_task
         not_default_collection_run_id,
     ]
 
-    for proj, collection_run_id in zip(projects, collection_run_id_project_list):
+    running_state_project_list = [
+        RunningState.SUCCESS,
+        RunningState.PENDING,
+        RunningState.SUCCESS,
+    ]
+
+    for proj, collection_run_id, running_state in zip(
+        projects,
+        collection_run_id_project_list,
+        running_state_project_list,
+        strict=True,
+    ):
         await create_pipeline(
             project_id=f"{proj.uuid}",
             dag_adjacency_list=fake_workbench_adjacency,
         )
         await create_tasks_from_project(
-            user=user, project=proj, state=StateType.PUBLISHED, progress=None
+            user=user, project=proj, state=running_state, progress=None
         )
         run = await create_comp_run(
             user=user,
             project=proj,
-            result=RunningState.SUCCESS,
+            result=running_state,
             started=datetime.now(tz=UTC) - timedelta(minutes=120),
             ended=datetime.now(tz=UTC) - timedelta(minutes=100),
             iteration=1,
@@ -317,3 +328,16 @@ async def test_rpc_list_computation_collection_runs_page_and_collection_run_task
     assert output.total == 4
     assert len(output.items) == 4
     isinstance(output, ComputationCollectionRunTaskRpcGetPage)
+
+    # Test filtering only running collection runs
+    output = await rpc_computations.list_computation_collection_runs_page(
+        rpc_client,
+        product_name="osparc",
+        user_id=user["id"],
+        project_ids=None,
+        filter_only_running=True,  # <-- This is the tested filter
+    )
+    assert output.total == 1
+    assert len(output.items) == 1
+    assert isinstance(output, ComputationCollectionRunRpcGetPage)
+    assert len(output.items[0].project_ids) == 2

services/director-v2/tests/unit/with_dbs/comp_scheduler/test_db_repositories_comp_runs.py

@@ -1019,7 +1019,7 @@ async def test_list_group_by_collection_run_id_with_project_filter(
     total_count, items = await repo.list_group_by_collection_run_id(
         product_name=run_metadata.get("product_name"),
         user_id=published_project_1.user["id"],
-        project_ids=[
+        project_ids_or_none=[
             published_project_1.project.uuid,
             published_project_2.project.uuid,
         ],