rerouted update_projects_networks

Author: Andrei Neagu

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/dynamic_scheduler/services.py

@@ -93,3 +93,16 @@ async def stop_dynamic_service(
         timeout_s=timeout_s,
     )
     assert result is None  # nosec
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def update_projects_networks(
+    rabbitmq_rpc_client: RabbitMQRPCClient, *, project_id: ProjectID
+) -> None:
+    result = await rabbitmq_rpc_client.request(
+        DYNAMIC_SCHEDULER_RPC_NAMESPACE,
+        _RPC_METHOD_NAME_ADAPTER.validate_python("update_projects_networks"),
+        project_id=project_id,
+        timeout_s=_RPC_DEFAULT_TIMEOUT_S,
+    )
+    assert result is None  # nosec

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/api/rpc/_services.py

@@ -56,3 +56,8 @@ async def stop_dynamic_service(
     return await scheduler_interface.stop_dynamic_service(
         app, dynamic_service_stop=dynamic_service_stop
     )
+
+
+@router.expose()
+async def update_projects_networks(app: FastAPI, *, project_id: ProjectID) -> None:
+    await scheduler_interface.update_projects_networks(app, project_id=project_id)

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/director_v2/_public_client.py

@@ -108,6 +108,9 @@ async def list_tracked_dynamic_services(
         )
         return TypeAdapter(list[DynamicServiceGet]).validate_python(response.json())
 
+    async def update_projects_networks(self, *, project_id: ProjectID) -> None:
+        await self.thin_client.patch_projects_networks(project_id=project_id)
+
 
 def setup_director_v2(app: FastAPI) -> None:
     public_client = DirectorV2Client(app)

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/director_v2/_thin_client.py

@@ -124,3 +124,10 @@ async def get_dynamic_services(
             "/dynamic_services",
             params=as_dict_exclude_unset(user_id=user_id, project_id=project_id),
         )
+
+    @retry_on_errors()
+    @expect_status(status.HTTP_204_NO_CONTENT)
+    async def patch_projects_networks(self, *, project_id: ProjectID) -> Response:
+        return await self.client.patch(
+            f"/dynamic_services/projects/{project_id}/-/networks"
+        )

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/scheduler_interface.py

@@ -73,3 +73,12 @@ async def stop_dynamic_service(
     )
 
     await set_request_as_stopped(app, dynamic_service_stop)
+
+
+async def update_projects_networks(app: FastAPI, *, project_id: ProjectID) -> None:
+    settings: ApplicationSettings = app.state.settings
+    if settings.DYNAMIC_SCHEDULER_USE_INTERNAL_SCHEDULER:
+        raise NotImplementedError
+
+    director_v2_client = DirectorV2Client.get_from_app_state(app)
+    await director_v2_client.update_projects_networks(project_id=project_id)

services/dynamic-scheduler/tests/unit/api_rpc/test_api_rpc__services.py

@@ -490,3 +490,24 @@ async def test_stop_dynamic_service_serializes_generic_errors(
             ),
             timeout_s=5,
         )
+
+
+@pytest.fixture
+def mock_director_v2_update_projects_networks(project_id: ProjectID) -> Iterator[None]:
+    with respx.mock(
+        base_url="http://director-v2:8000/v2",
+        assert_all_called=False,
+        assert_all_mocked=True,  # IMPORTANT: KEEP always True!
+    ) as mock:
+        mock.patch(f"/dynamic_services/projects/{project_id}/-/networks").respond(
+            status.HTTP_204_NO_CONTENT
+        )
+        yield None
+
+
+async def test_update_projects_networks(
+    mock_director_v2_update_projects_networks: None,
+    rpc_client: RabbitMQRPCClient,
+    project_id: ProjectID,
+):
+    await services.update_projects_networks(rpc_client, project_id=project_id)

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

@@ -83,19 +83,6 @@ async def restart_dynamic_service(app: web.Application, node_uuid: str) -> None:
     )
 
 
-@log_decorator(logger=_log)
-async def update_dynamic_service_networks_in_project(
-    app: web.Application, project_id: ProjectID
-) -> None:
-    settings: DirectorV2Settings = get_plugin_settings(app)
-    backend_url = (
-        URL(settings.base_url) / f"dynamic_services/projects/{project_id}/-/networks"
-    )
-    await request_director_v2(
-        app, "PATCH", backend_url, expected_status=web.HTTPNoContent
-    )
-
-
 @log_decorator(logger=_log)
 async def get_project_inactivity(
     app: web.Application,

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

@@ -21,7 +21,6 @@
     request_retrieve_dyn_service,
     restart_dynamic_service,
     retrieve,
-    update_dynamic_service_networks_in_project,
 )
 from ._core_utils import is_healthy
 from .exceptions import DirectorServiceError
@@ -43,6 +42,5 @@
     "retrieve",
     "set_project_run_policy",
     "stop_pipeline",
-    "update_dynamic_service_networks_in_project",
 )
 # nopycln: file

services/web/server/src/simcore_service_webserver/dynamic_scheduler/api.py

@@ -148,3 +148,11 @@ async def stop_dynamic_services_in_project(
         ]
 
         await logged_gather(*services_to_stop)
+
+
+async def update_projects_networks(
+    app: web.Application, *, project_id: ProjectID
+) -> None:
+    await services.update_projects_networks(
+        get_rabbitmq_rpc_client(app), project_id=project_id
+    )

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

@@ -26,7 +26,8 @@
 
 from ..application_settings import get_application_settings
 from ..catalog import client as catalog_client
-from ..director_v2 import api
+from ..director_v2 import api as director_v2_api
+from ..dynamic_scheduler import api as dynamic_scheduler_api
 from ..folders import _folders_db as folders_db
 from ..storage.api import (
     copy_data_folders_from_project,
@@ -376,13 +377,13 @@ async def create_project(  # pylint: disable=too-many-arguments,too-many-branche
             await db.set_hidden_flag(new_project["uuid"], hidden=False)
 
         # update the network information in director-v2
-        await api.update_dynamic_service_networks_in_project(
-            request.app, ProjectID(new_project["uuid"])
+        await dynamic_scheduler_api.update_projects_networks(
+            request.app, project_id=ProjectID(new_project["uuid"])
         )
         task_progress.update()
 
         # This is a new project and every new graph needs to be reflected in the pipeline tables
-        await api.create_or_update_pipeline(
+        await director_v2_api.create_or_update_pipeline(
             request.app, user_id, new_project["uuid"], product_name
         )
         # get the latest state of the project (lastChangeDate for instance)