fix

Author: matusdrobuliak66

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

@@ -35,7 +35,6 @@
     sa.Column(
         "project_id",
         sa.String,
-        sa.ForeignKey("comp_pipeline.project_id"),
         doc="Project that contains the node associated to this task",
     ),
     sa.Column("node_id", sa.String, doc="Node associated to this task"),

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

@@ -1,5 +1,4 @@
 import sqlalchemy as sa
-from models_library.projects_nodes_io import NodeID
 from pydantic import PositiveInt
 from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine
 
@@ -38,7 +37,7 @@ async def update_for_run_id_and_node_id(
     conn: AsyncConnection | None = None,
     *,
     run_id: PositiveInt,
-    node_id: NodeID,
+    node_id: str,
     data: dict,
 ):
     async with pass_or_acquire_connection(engine, connection=conn) as _conn:
@@ -50,7 +49,7 @@ async def update_for_run_id_and_node_id(
             )
             .where(
                 (comp_run_snapshot_tasks.c.run_id == run_id)
-                & (comp_run_snapshot_tasks.c.node_id == f"{node_id}")
+                & (comp_run_snapshot_tasks.c.node_id == node_id)
             )
             .returning(*COMP_RUN_SNAPSHOT_TASKS_DB_COLS)
         )

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

@@ -1,5 +1,4 @@
 import sqlalchemy as sa
-from models_library.projects import ProjectID
 from pydantic import PositiveInt
 from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine
 
@@ -11,7 +10,7 @@ async def get_latest_run_id_for_project(
     engine: AsyncEngine,
     conn: AsyncConnection | None = None,
     *,
-    project_id: ProjectID,
+    project_id: str,
 ) -> PositiveInt:
     # Get latest run per (project_uuid, user_id)
     project_and_user_latest_runs = (
@@ -21,7 +20,7 @@ async def get_latest_run_id_for_project(
             sa.func.max(comp_runs.c.iteration).label("latest_iteration"),
             sa.func.max(comp_runs.c.created).label("created"),
         )
-        .where(comp_runs.c.project_uuid == f"{project_id}")
+        .where(comp_runs.c.project_uuid == project_id)
         .group_by(comp_runs.c.project_uuid, comp_runs.c.user_id)
         .subquery("project_and_user_latest_runs")
     )

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

@@ -3,7 +3,6 @@
 import sqlalchemy as sa
 from common_library.json_serialization import json_dumps, json_loads
 from models_library.projects import ProjectID
-from models_library.projects_nodes_io import NodeID
 from models_library.users import UserID
 from pydantic import TypeAdapter
 from servicelib.db_asyncpg_utils import create_async_engine_and_database_ready
@@ -114,15 +113,15 @@ async def write_ports_configuration(
             )
             # 2. Get latest run id for the project
             _latest_run_id = await get_latest_run_id_for_project(
-                engine, connection, project_id=ProjectID(project_id)
+                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=NodeID(node_uuid),
+                node_id=node_uuid,
                 data={
                     "schema": node_configuration["schema"],
                     "inputs": node_configuration["inputs"],

services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/_scheduler_dask.py

@@ -117,7 +117,6 @@ async def _start_tasks(
                     client.send_computation_tasks(
                         user_id=user_id,
                         project_id=project_id,
-                        run_id=comp_run.run_id,
                         tasks={node_id: task.image},
                         hardware_info=task.hardware_info,
                         callback=wake_up_callback,

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

@@ -358,7 +358,7 @@ async def get_latest_run_id_for_project(
         project_id: ProjectID,
     ) -> PositiveInt:
         return await get_latest_run_id_for_project(
-            self.db_engine, conn, project_id=project_id
+            self.db_engine, conn, project_id=f"{project_id}"
         )
 
     async def create(

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

@@ -27,14 +27,22 @@ async def batch_create(
     ) -> list[CompRunSnapshotTaskAtDBGet]:
         async with transaction_context(self.db_engine) as conn:
 
-            result = 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]
+            try:
+                result = 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]
+            except Exception as e:
+                logger.error(
+                    "Failed to batch create comp run snapshot tasks: %s",
+                    e,
+                    exc_info=True,
+                )
+                raise
 
     async def update_for_run_id_and_node_id(
         self,
@@ -45,6 +53,10 @@ async def update_for_run_id_and_node_id(
         data: dict,
     ) -> CompRunSnapshotTaskAtDBGet:
         row = await update_for_run_id_and_node_id(
-            self.db_engine, conn=connection, run_id=run_id, node_id=node_id, data=data
+            self.db_engine,
+            conn=connection,
+            run_id=run_id,
+            node_id=f"{node_id}",
+            data=data,
         )
         return CompRunSnapshotTaskAtDBGet.model_validate(row)