Merge branch 'master' into enh/detect-new-version

Author: odeimaiz

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/aiohttp/monitoring.py

@@ -25,7 +25,7 @@
 )
 from .typing_extension import Handler
 
-log = logging.getLogger(__name__)
+_logger = logging.getLogger(__name__)
 
 _PROMETHEUS_METRICS: Final[str] = f"{__name__}.prometheus_metrics"  # noqa: N816
 
@@ -59,7 +59,6 @@ def middleware_factory(
     async def middleware_handler(request: web.Request, handler: Handler):
         # See https://prometheus.io/docs/concepts/metric_types
 
-        log_exception: BaseException | None = None
         response: web.StreamResponse = web.HTTPInternalServerError()
 
         canonical_endpoint = request.path
@@ -68,7 +67,7 @@ async def middleware_handler(request: web.Request, handler: Handler):
         start_time = perf_counter()
         try:
             if enter_middleware_cb:
-                with log_catch(logger=log, reraise=False):
+                with log_catch(logger=_logger, reraise=False):
                     await enter_middleware_cb(request)
 
             metrics = request.app[_PROMETHEUS_METRICS]
@@ -92,16 +91,10 @@ async def middleware_handler(request: web.Request, handler: Handler):
 
         except web.HTTPServerError as exc:
             response = exc
-            log_exception = exc
             raise
 
         except web.HTTPException as exc:
             response = exc
-            log_exception = None
-            raise
-
-        except Exception as exc:  # pylint: disable=broad-except
-            log_exception = exc
             raise
 
         finally:
@@ -117,23 +110,9 @@ async def middleware_handler(request: web.Request, handler: Handler):
             )
 
             if exit_middleware_cb:
-                with log_catch(logger=log, reraise=False):
+                with log_catch(logger=_logger, reraise=False):
                     await exit_middleware_cb(request, response)
 
-            if log_exception:
-                log.error(
-                    'Unexpected server error "%s" from access: %s "%s %s" done '
-                    "in %3.2f secs. Responding with status %s",
-                    type(log_exception),
-                    request.remote,
-                    request.method,
-                    request.path,
-                    response_latency_seconds,
-                    response.status,
-                    exc_info=log_exception,
-                    stack_info=True,
-                )
-
         return response
 
     setattr(  # noqa: B010

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/catalog/src/simcore_service_catalog/clients/director.py

@@ -82,7 +82,9 @@ def _unenvelope_or_raise_error(
         """
         body = resp.json()
 
-        assert "data" in body or "error" in body  # nosec
+        assert (
+            "data" in body or "error" in body
+        ), f"here is the failing {body=}, {resp.request=}"  # nosec
         data = body.get("data")
         error = body.get("error")
 
@@ -250,7 +252,6 @@ async def get_service_extras(
                     # discrete resources (custom made ones) ---
                     # check if the service requires GPU support
                     if not invalid_with_msg and _validate_kind(entry, "VRAM"):
-
                         result["node_requirements"]["GPU"] = 1
                     if not invalid_with_msg and _validate_kind(entry, "MPI"):
                         result["node_requirements"]["MPI"] = 1
@@ -282,7 +283,6 @@ async def get_service_extras(
             for dl, sl in _ORG_LABELS_TO_SCHEMA_LABELS.items()
             if dl in labels
         }:
-
             result.update({"service_build_details": service_build_details})
 
         return TypeAdapter(ServiceExtras).validate_python(result)

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/director/src/simcore_service_director/core/application.py

@@ -4,6 +4,7 @@
 from fastapi import FastAPI
 from servicelib.async_utils import cancel_sequential_workers
 from servicelib.fastapi.client_session import setup_client_session
+from servicelib.fastapi.http_error import set_app_default_http_error_handlers
 from servicelib.fastapi.tracing import (
     initialize_fastapi_app_tracing,
     setup_tracing,
@@ -73,6 +74,7 @@ def create_app(settings: ApplicationSettings) -> FastAPI:
         initialize_fastapi_app_tracing(app)
 
     # ERROR HANDLERS
+    set_app_default_http_error_handlers(app)
 
     # EVENTS
     async def _on_startup() -> None: