fix

Author: matusdrobuliak66

packages/postgres-database/src/simcore_postgres_database/utils_comp_run_snapshot_tasks.py

@@ -56,4 +56,5 @@ async def update_for_run_id_and_node_id(
         row = await result.one_or_none()
         if row is None:
             raise ValueError("improve message")
+            # return None
         return row

packages/simcore-sdk/src/simcore_sdk/node_ports_common/dbmanager.py

@@ -7,7 +7,7 @@
 from pydantic import TypeAdapter
 from servicelib.db_asyncpg_utils import create_async_engine_and_database_ready
 from settings_library.node_ports import NodePortsSettings
-from simcore_postgres_database.models.comp_tasks import comp_tasks
+from simcore_postgres_database.models.comp_tasks import NodeClass, comp_tasks
 from simcore_postgres_database.models.projects import projects
 from simcore_postgres_database.utils_comp_run_snapshot_tasks import (
     update_for_run_id_and_node_id,
@@ -111,24 +111,28 @@ async def write_ports_configuration(
                     run_hash=node_configuration.get("run_hash"),
                 )
             )
-            # 2. Get latest run id for the project
-            _latest_run_id = await get_latest_run_id_for_project(
-                engine, connection, project_id=project_id
-            )
 
-            # 3. Update comp_run_snapshot_tasks table
-            await update_for_run_id_and_node_id(
-                engine,
-                connection,
-                run_id=_latest_run_id,
-                node_id=node_uuid,
-                data={
-                    "schema": node_configuration["schema"],
-                    "inputs": node_configuration["inputs"],
-                    "outputs": node_configuration["outputs"],
-                    "run_hash": node_configuration.get("run_hash"),
-                },
-            )
+            # 2. Update comp_run_snapshot_tasks table only if the node is computational
+            node = await _get_node_from_db(project_id, node_uuid, connection)
+            if node.node_class == NodeClass.COMPUTATIONAL.value:
+                # 2.1 Get latest run id for the project
+                _latest_run_id = await get_latest_run_id_for_project(
+                    engine, connection, project_id=project_id
+                )
+
+                # 2.2 Update comp_run_snapshot_tasks table
+                await update_for_run_id_and_node_id(
+                    engine,
+                    connection,
+                    run_id=_latest_run_id,
+                    node_id=node_uuid,
+                    data={
+                        "schema": node_configuration["schema"],
+                        "inputs": node_configuration["inputs"],
+                        "outputs": node_configuration["outputs"],
+                        "run_hash": node_configuration.get("run_hash"),
+                    },
+                )
 
     async def get_ports_configuration_from_node_uuid(
         self, project_id: str, node_uuid: str

services/director-v2/src/simcore_service_director_v2/models/comp_run_snapshot_tasks.py

@@ -1,6 +1,7 @@
+from models_library.resource_tracker import HardwareInfo
 from pydantic import ConfigDict, PositiveInt
 
-from .comp_tasks import BaseCompTaskAtDB, HardwareInfo, Image
+from .comp_tasks import BaseCompTaskAtDB, Image
 
 
 class CompRunSnapshotTaskAtDBGet(BaseCompTaskAtDB):

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

@@ -1,18 +1,11 @@
 import logging
 
-from models_library.projects_nodes_io import NodeID
-from pydantic import PositiveInt
 from simcore_postgres_database.utils_comp_run_snapshot_tasks import (
     COMP_RUN_SNAPSHOT_TASKS_DB_COLS,
-    update_for_run_id_and_node_id,
 )
 from simcore_postgres_database.utils_repos import (
     transaction_context,
 )
-from simcore_service_director_v2.models.comp_run_snapshot_tasks import (
-    CompRunSnapshotTaskAtDBGet,
-)
-from sqlalchemy.ext.asyncio import AsyncConnection
 
 from ..tables import comp_run_snapshot_tasks
 from ._base import BaseRepository
@@ -24,18 +17,18 @@ class CompRunsSnapshotTasksRepository(BaseRepository):
 
     async def batch_create(
         self, *, data: list[dict]
-    ) -> list[CompRunSnapshotTaskAtDBGet]:
+    ) -> None:  # list[CompRunSnapshotTaskAtDBGet]:
         async with transaction_context(self.db_engine) as conn:
 
             try:
-                result = await conn.execute(
+                await conn.execute(
                     comp_run_snapshot_tasks.insert().returning(
                         *COMP_RUN_SNAPSHOT_TASKS_DB_COLS
                     ),
                     data,
                 )
-                rows = result.fetchall()
-                return [CompRunSnapshotTaskAtDBGet.model_validate(row) for row in rows]
+                # rows = result.fetchall()
+                # return [CompRunSnapshotTaskAtDBGet.model_validate(row) for row in rows]
             except Exception as e:
                 logger.error(
                     "Failed to batch create comp run snapshot tasks: %s",
@@ -44,19 +37,19 @@ async def batch_create(
                 )
                 raise
 
-    async def update_for_run_id_and_node_id(
-        self,
-        connection: AsyncConnection | None = None,
-        *,
-        run_id: PositiveInt,
-        node_id: NodeID,
-        data: dict,
-    ) -> CompRunSnapshotTaskAtDBGet:
-        row = await update_for_run_id_and_node_id(
-            self.db_engine,
-            conn=connection,
-            run_id=run_id,
-            node_id=f"{node_id}",
-            data=data,
-        )
-        return CompRunSnapshotTaskAtDBGet.model_validate(row)
+    # async def update_for_run_id_and_node_id(
+    #     self,
+    #     connection: AsyncConnection | None = None,
+    #     *,
+    #     run_id: PositiveInt,
+    #     node_id: NodeID,
+    #     data: dict,
+    # ) -> CompRunSnapshotTaskAtDBGet:
+    #     row = await update_for_run_id_and_node_id(
+    #         self.db_engine,
+    #         conn=connection,
+    #         run_id=run_id,
+    #         node_id=f"{node_id}",
+    #         data=data,
+    #     )
+    #     return CompRunSnapshotTaskAtDBGet.model_validate(row)