add RPC communication

Author: giancarloromeo

packages/models-library/src/models_library/rpc/webserver/projects/__init__.py

@@ -0,0 +1,7 @@
+from pydantic import BaseModel
+
+
+class ProjectNodeGet(BaseModel):
+    key: str
+    version: str
+    label: str

packages/models-library/src/models_library/rpc/webserver/projects/projects_nodes.py

@@ -0,0 +1,26 @@
+import logging
+
+from models_library.api_schemas_webserver import WEBSERVER_RPC_NAMESPACE
+from models_library.api_schemas_webserver.projects import ProjectID
+from models_library.projects_nodes import Node
+from models_library.rabbitmq_basic_types import RPCMethodName
+from models_library.rpc.webserver.projects.projects_nodes import ProjectNodeGet
+from pydantic import TypeAdapter
+from servicelib.logging_utils import log_decorator
+from servicelib.rabbitmq import RabbitMQRPCClient
+
+_logger = logging.getLogger(__name__)
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def list_project_nodes(
+    rpc_client: RabbitMQRPCClient,
+    *,
+    project_uuid: ProjectID,
+) -> list[Node]:
+    result: list[ProjectNodeGet] = await rpc_client.request(
+        WEBSERVER_RPC_NAMESPACE,
+        TypeAdapter(RPCMethodName).validate_python("list_project_nodes"),
+        project_uuid=project_uuid,
+    )
+    return TypeAdapter(list[Node]).validate_python(result)

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/webserver/projects/projects_nodes.py

@@ -37,6 +37,9 @@
 from servicelib.async_utils import run_sequentially_in_context
 from servicelib.logging_utils import log_decorator
 from servicelib.rabbitmq import RabbitMQRPCClient
+from servicelib.rabbitmq.rpc_interfaces.webserver.projects import (
+    projects_nodes as projects_nodes_rpc,
+)
 from simcore_postgres_database.utils_projects_metadata import DBProjectNotFoundError
 from starlette import status
 from starlette.requests import Request
@@ -301,9 +304,14 @@ async def create_computation(  # noqa: PLR0913 # pylint: disable=too-many-positi
 
         # check if current state allow to modify the computation
         await _check_pipeline_not_running_or_raise_409(comp_tasks_repo, computation)
-
+        workbench = {
+            node.key: node
+            for node in await projects_nodes_rpc.list_project_nodes(
+                rpc_client, project_uuid=computation.project_id
+            )
+        }
         # create the complete DAG graph
-        complete_dag = create_complete_dag(project.workbench)
+        complete_dag = create_complete_dag(workbench)
         # find the minimal viable graph to be run
         minimal_computational_dag: nx.DiGraph = (
             await create_minimal_computational_graph_based_on_selection(

services/director-v2/src/simcore_service_director_v2/api/routes/computations.py

@@ -42,9 +42,9 @@ def create_complete_dag(workbench: NodesDict) -> nx.DiGraph:
         )
         if node.input_nodes:
             for input_node_id in node.input_nodes:
-                predecessor_node = workbench.get(NodeIDStr(input_node_id))
+                predecessor_node = workbench.get(f"{input_node_id}")
                 if predecessor_node:
-                    dag_graph.add_edge(str(input_node_id), node_id)
+                    dag_graph.add_edge(f"{input_node_id}", node_id)
 
     return dag_graph
 

services/director-v2/src/simcore_service_director_v2/utils/dags.py

@@ -36,16 +36,16 @@
 ]
 
 
-async def get_node(
+async def get_project_node(
     app: web.Application,
     connection: AsyncConnection | None = None,
     *,
-    project_id: ProjectID,
+    project_uuid: ProjectID,
     node_id: NodeID,
 ) -> Node:
     async with transaction_context(get_asyncpg_engine(app), connection) as conn:
         get_stmt = sa.select(*_SELECTION_PROJECTS_NODES_DB_ARGS).where(
-            (projects_nodes.c.project_uuid == f"{project_id}")
+            (projects_nodes.c.project_uuid == f"{project_uuid}")
             & (projects_nodes.c.node_id == f"{node_id}")
         )
 
@@ -55,44 +55,48 @@ async def get_node(
         row = await result.first()
         if row is None:
             raise NodeNotFoundError(
-                project_uuid=f"{project_id}", node_uuid=f"{node_id}"
+                project_uuid=f"{project_uuid}", node_uuid=f"{node_id}"
             )
         assert row  # nosec
         return Node.model_validate(row, from_attributes=True)
 
 
-async def list_nodes(
+async def list_project_nodes(
     app: web.Application,
     connection: AsyncConnection | None = None,
     *,
-    project_id: ProjectID,
+    project_uuid: ProjectID,
 ) -> list[Node]:
     async with transaction_context(get_asyncpg_engine(app), connection) as conn:
         result = await conn.stream(
             sa.select(*_SELECTION_PROJECTS_NODES_DB_ARGS).where(
-                projects_nodes.c.project_uuid == f"{project_id}"
+                projects_nodes.c.project_uuid == f"{project_uuid}"
             )
         )
         rows = await result.all() or []
         return TypeAdapter(list[Node]).validate_python(rows)
 
 
-async def update_node(
+async def update_project_node(
     app: web.Application,
     connection: AsyncConnection | None = None,
     *,
-    project_id: ProjectID,
+    project_uuid: ProjectID,
     node_id: NodeID,
     partial_node: PartialNode,
-) -> None:
+) -> Node:
     values = partial_node.model_dump(mode="json", exclude_unset=True)
 
     async with transaction_context(get_asyncpg_engine(app), connection) as conn:
-        await conn.stream(
+        result = await conn.stream(
             projects_nodes.update()
             .values(**values)
             .where(
-                (projects_nodes.c.project_uuid == f"{project_id}")
+                (projects_nodes.c.project_uuid == f"{project_uuid}")
                 & (projects_nodes.c.node_id == f"{node_id}")
             )
+            .returning(*_SELECTION_PROJECTS_NODES_DB_ARGS)
         )
+        row = await result.first()
+        assert row
+        return Node.model_validate(row, from_attributes=True)

services/web/server/src/simcore_service_webserver/projects/_projects_nodes_repository.py

@@ -0,0 +1,27 @@
+from aiohttp import web
+from models_library.api_schemas_webserver import WEBSERVER_RPC_NAMESPACE
+from models_library.projects import ProjectID
+from models_library.projects_nodes import Node
+from models_library.rpc.webserver.projects.projects_nodes import ProjectNodeGet
+from pydantic import TypeAdapter
+from servicelib.rabbitmq import RPCRouter
+
+from ....rabbitmq import get_rabbitmq_rpc_server
+from ... import projects_service
+
+router = RPCRouter()
+
+
+@router.expose()
+async def list_project_nodes(
+    app: web.Application,
+    *,
+    project_uuid: ProjectID,
+) -> list[ProjectNodeGet]:
+    nodes: list[Node] = await projects_service.list_project_nodes(app, project_uuid)
+    return TypeAdapter(list[ProjectNodeGet]).validate_python(nodes)
+
+
+async def register_rpc_routes(app: web.Application):
+    rpc_server = get_rabbitmq_rpc_server(app)
+    await rpc_server.register_router(router, WEBSERVER_RPC_NAMESPACE, app)

services/web/server/src/simcore_service_webserver/projects/api/__init__.py

@@ -9,6 +9,7 @@
 from servicelib.aiohttp.application_setup import ModuleCategory, app_module_setup
 
 from .._constants import APP_SETTINGS_KEY
+from ..rabbitmq import setup_rabbitmq
 from . import (
     _comments_handlers,
     _crud_handlers,
@@ -26,6 +27,7 @@
 )
 from ._observer import setup_project_observer_events
 from ._projects_access import setup_projects_access
+from .api.rpc.routes import register_rpc_routes
 from .db import setup_projects_db
 
 logger = logging.getLogger(__name__)
@@ -64,4 +66,9 @@ def setup_projects(app: web.Application) -> bool:
     app.router.add_routes(_workspaces_handlers.routes)
     app.router.add_routes(_trash_rest.routes)
 
+    # RPC
+    setup_rabbitmq(app)
+    if app[APP_SETTINGS_KEY].WEBSERVER_RABBITMQ:
+        app.on_startup.append(register_rpc_routes)
+
     return True