added new project_networks

Author: Andrei Neagu

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/core/events.py

@@ -6,13 +6,17 @@
     create_remote_docker_client_input_state,
     remote_docker_client_lifespan,
 )
+from servicelib.fastapi.postgres_lifespan import (
+    create_postgres_database_input_state,
+)
 from servicelib.fastapi.prometheus_instrumentation import (
     create_prometheus_instrumentationmain_input_state,
     prometheus_instrumentation_lifespan,
 )
 
 from .._meta import APP_FINISHED_BANNER_MSG, APP_STARTED_BANNER_MSG
 from ..api.rpc.routes import rpc_api_routes_lifespan
+from ..repository.events import repository_lifespan_manager
 from ..services.catalog import catalog_lifespan
 from ..services.deferred_manager import deferred_manager_lifespan
 from ..services.director_v0 import director_v0_lifespan
@@ -36,6 +40,7 @@ async def _settings_lifespan(app: FastAPI) -> AsyncIterator[State]:
     settings: ApplicationSettings = app.state.settings
 
     yield {
+        **create_postgres_database_input_state(settings.DYNAMIC_SCHEDULER_POSTGRES),
         **create_prometheus_instrumentationmain_input_state(
             enabled=settings.DYNAMIC_SCHEDULER_PROMETHEUS_INSTRUMENTATION_ENABLED
         ),
@@ -49,6 +54,7 @@ def create_app_lifespan() -> LifespanManager:
     app_lifespan = LifespanManager()
     app_lifespan.add(_settings_lifespan)
 
+    app_lifespan.include(repository_lifespan_manager)
     app_lifespan.add(director_v2_lifespan)
     app_lifespan.add(director_v0_lifespan)
     app_lifespan.add(catalog_lifespan)

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/repository/__init__.py

@@ -0,0 +1,26 @@
+import logging
+from collections.abc import AsyncIterator
+
+from fastapi import FastAPI
+from fastapi_lifespan_manager import LifespanManager, State
+from servicelib.fastapi.postgres_lifespan import (
+    PostgresLifespanState,
+    postgres_database_lifespan,
+)
+
+_logger = logging.getLogger(__name__)
+
+
+async def _database_lifespan(app: FastAPI, state: State) -> AsyncIterator[State]:
+    app.state.engine = state[PostgresLifespanState.POSTGRES_ASYNC_ENGINE]
+
+    # TODO initialize all the repos here?
+
+    # app.state.default_product_name = await repo.get_default_product_name()
+
+    yield {}
+
+
+repository_lifespan_manager = LifespanManager()
+repository_lifespan_manager.add(postgres_database_lifespan)
+repository_lifespan_manager.add(_database_lifespan)

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/repository/events.py

@@ -0,0 +1,57 @@
+import sqlalchemy as sa
+from common_library.errors_classes import OsparcErrorMixin
+from models_library.projects import ProjectID
+from models_library.projects_networks import NetworksWithAliases, ProjectsNetworks
+from simcore_postgres_database.models.projects_networks import projects_networks
+from simcore_postgres_database.utils_repos import (
+    pass_or_acquire_connection,
+    transaction_context,
+)
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine
+
+
+class BaseProjectNetwroksError(OsparcErrorMixin, RuntimeError):
+    msg_template: str = "project networks unexpected error"
+
+
+class ProjectNetworkNotFoundError(BaseProjectNetwroksError):
+    msg_template: str = "no networks found for project {project_id}"
+
+
+class ProjectNetworksRepo:
+    def __init__(self, engine: AsyncEngine):
+        self.engine = engine
+
+    async def get_projects_networks(
+        self, connection: AsyncConnection | None = None, *, project_id: ProjectID
+    ) -> ProjectsNetworks:
+        async with pass_or_acquire_connection(self.engine, connection) as conn:
+            result = await conn.execute(
+                sa.select(projects_networks).where(
+                    projects_networks.c.project_uuid == f"{project_id}"
+                )
+            )
+            row = result.first()
+        if not row:
+            raise ProjectNetworkNotFoundError(project_id=project_id)
+        return ProjectsNetworks.model_validate(row)
+
+    async def upsert_projects_networks(
+        self,
+        connection: AsyncConnection | None = None,
+        *,
+        project_id: ProjectID,
+        networks_with_aliases: NetworksWithAliases,
+    ) -> None:
+        projects_networks_to_insert = ProjectsNetworks.model_validate(
+            {"project_uuid": project_id, "networks_with_aliases": networks_with_aliases}
+        )
+
+        async with transaction_context(self.engine, connection) as conn:
+            row_data = projects_networks_to_insert.model_dump(mode="json")
+            insert_stmt = pg_insert(projects_networks).values(**row_data)
+            upsert_snapshot = insert_stmt.on_conflict_do_update(
+                index_elements=[projects_networks.c.project_uuid], set_=row_data
+            )
+            await conn.execute(upsert_snapshot)