🎨 Add filter to show only running jobs in Activity Overview (#8055)

Author: matusdrobuliak66

packages/postgres-database/src/simcore_postgres_database/migration/versions/61b98a60e934_computational_collection_uniquencess.py

@@ -0,0 +1,35 @@
+"""computational collection uniquencess
+
+Revision ID: 61b98a60e934
+Revises: df61d1b2b967
+Create Date: 2025-07-08 15:40:12.714684+00:00
+
+"""
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "61b98a60e934"
+down_revision = "df61d1b2b967"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_unique_constraint(
+        "client_or_system_generated_id_uniqueness",
+        "comp_runs_collections",
+        ["client_or_system_generated_id"],
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(
+        "client_or_system_generated_id_uniqueness",
+        "comp_runs_collections",
+        type_="unique",
+    )
+    # ### end Alembic commands ###

packages/postgres-database/src/simcore_postgres_database/models/comp_runs_collections.py

@@ -35,4 +35,7 @@
         "ix_comp_runs_collections_client_or_system_generated_id",
         "client_or_system_generated_id",
     ),
+    sa.UniqueConstraint(
+        "client_or_system_generated_id", name="client_or_system_generated_id_uniqueness"
+    ),
 )

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,
         ],

services/static-webserver/client/source/class/osparc/data/Resources.js

@@ -368,9 +368,7 @@ qx.Class.define("osparc.data.Resources", {
         endpoints: {
           getPageLatest: {
             method: "GET",
-            // uncomment this when the backend supports filtering by runningOnly #8055
-            // url: statics.API + "/computation-collection-runs?offset={offset}&limit={limit}&order_by={orderBy}&filter_only_running={runningOnly}"
-            url: statics.API + "/computation-collection-runs?offset={offset}&limit={limit}&order_by={orderBy}&filter_only_running=false"
+            url: statics.API + "/computation-collection-runs?offset={offset}&limit={limit}&order_by={orderBy}&filter_only_running={runningOnly}"
           },
           getPageHistory: {
             method: "GET",

services/web/server/src/simcore_service_webserver/director_v2/_comp_runs_collections_repository.py

@@ -5,6 +5,7 @@
 from pydantic import TypeAdapter
 from simcore_postgres_database.models.comp_runs_collections import comp_runs_collections
 from sqlalchemy import func
+from sqlalchemy.dialects.postgresql import insert as pg_insert
 
 from ._comp_runs_collections_models import CompRunCollectionDBGet
 
@@ -63,3 +64,30 @@ async def get_comp_run_collection_or_none_by_client_generated_id(
     if row is None:
         return None
     return CompRunCollectionDBGet.model_validate(row)
+
+
+async def upsert_comp_run_collection(
+    conn,
+    client_or_system_generated_id: str,
+    client_or_system_generated_display_name: str,
+    is_generated_by_system: bool,
+) -> CollectionRunID:
+    """Upsert a computational run collection. If it exists, only update the modified time."""
+    insert_stmt = pg_insert(comp_runs_collections).values(
+        client_or_system_generated_id=client_or_system_generated_id,
+        client_or_system_generated_display_name=client_or_system_generated_display_name,
+        is_generated_by_system=is_generated_by_system,
+        created=func.now(),
+        modified=func.now(),
+    )
+    on_update_stmt = insert_stmt.on_conflict_do_update(
+        index_elements=[comp_runs_collections.c.client_or_system_generated_id],
+        set_={
+            "modified": func.now(),
+        },
+    )
+    result = await conn.stream(
+        on_update_stmt.returning(comp_runs_collections.c.collection_run_id)
+    )
+    collection_id_tuple: tuple[UUID] = await result.one()
+    return TypeAdapter(CollectionRunID).validate_python(collection_id_tuple[0])

services/web/server/src/simcore_service_webserver/director_v2/_comp_runs_collections_service.py

@@ -27,6 +27,22 @@ async def create_comp_run_collection(
         )
 
 
+async def upsert_comp_run_collection(
+    app: web.Application,
+    *,
+    client_or_system_generated_id: str,
+    client_or_system_generated_display_name: str,
+    is_generated_by_system: bool,
+) -> CollectionRunID:
+    async with transaction_context(get_asyncpg_engine(app)) as conn:
+        return await _comp_runs_collections_repository.upsert_comp_run_collection(
+            conn=conn,
+            client_or_system_generated_id=client_or_system_generated_id,
+            client_or_system_generated_display_name=client_or_system_generated_display_name,
+            is_generated_by_system=is_generated_by_system,
+        )
+
+
 async def get_comp_run_collection_or_none_by_id(
     app: web.Application, *, collection_run_id: CollectionRunID
 ) -> CompRunCollectionDBGet | None: