✨ connect Postgres and add projects_networks to dynamic-scheduler ⚠️

packages/service-library/src/servicelib/logging_utils.py

@@ -415,15 +415,15 @@ def log_context(
     if extra:
         kwargs["extra"] = extra
     log_msg = f"Starting {msg} ..."
-    logger.log(level, log_msg, *args, **kwargs)
+    logger.log(level, log_msg, *args, **kwargs, stacklevel=3)
     yield
     duration = (
         f" in {(datetime.now() - start ).total_seconds()}s"  # noqa: DTZ005
         if log_duration
         else ""
     )
     log_msg = f"Finished {msg}{duration}"
-    logger.log(level, log_msg, *args, **kwargs)
+    logger.log(level, log_msg, *args, **kwargs, stacklevel=3)
 
 
 def guess_message_log_level(message: str) -> LogLevelInt:

services/docker-compose.yml

@@ -550,27 +550,38 @@ services:
     networks:
       - default
     environment:
-      LOG_FORMAT_LOCAL_DEV_ENABLED: ${LOG_FORMAT_LOCAL_DEV_ENABLED}
+      DIRECTOR_V2_HOST: ${DIRECTOR_V2_HOST}
+      DIRECTOR_V2_PORT: ${DIRECTOR_V2_PORT}
+
+      DYNAMIC_SCHEDULER_LOGLEVEL: ${DYNAMIC_SCHEDULER_LOGLEVEL}
+      DYNAMIC_SCHEDULER_PROFILING: ${DYNAMIC_SCHEDULER_PROFILING}
+      DYNAMIC_SCHEDULER_STOP_SERVICE_TIMEOUT: ${DYNAMIC_SCHEDULER_STOP_SERVICE_TIMEOUT}
+      DYNAMIC_SCHEDULER_TRACING: ${DYNAMIC_SCHEDULER_TRACING}
+      DYNAMIC_SCHEDULER_UI_STORAGE_SECRET: ${DYNAMIC_SCHEDULER_UI_STORAGE_SECRET}
+      DYNAMIC_SCHEDULER_USE_INTERNAL_SCHEDULER: ${DYNAMIC_SCHEDULER_USE_INTERNAL_SCHEDULER}
+      DYNAMIC_SIDECAR_API_SAVE_RESTORE_STATE_TIMEOUT: ${DYNAMIC_SIDECAR_API_SAVE_RESTORE_STATE_TIMEOUT}
+
       LOG_FILTER_MAPPING : ${LOG_FILTER_MAPPING}
+      LOG_FORMAT_LOCAL_DEV_ENABLED: ${LOG_FORMAT_LOCAL_DEV_ENABLED}
+
+      POSTGRES_DB: ${POSTGRES_DB}
+      POSTGRES_HOST: ${POSTGRES_HOST}
+      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
+      POSTGRES_PORT: ${POSTGRES_PORT}
+      POSTGRES_USER: ${POSTGRES_USER}
+
       RABBIT_HOST: ${RABBIT_HOST}
       RABBIT_PASSWORD: ${RABBIT_PASSWORD}
       RABBIT_PORT: ${RABBIT_PORT}
       RABBIT_SECURE: ${RABBIT_SECURE}
       RABBIT_USER: ${RABBIT_USER}
+
       REDIS_HOST: ${REDIS_HOST}
+      REDIS_PASSWORD: ${REDIS_PASSWORD}
       REDIS_PORT: ${REDIS_PORT}
       REDIS_SECURE: ${REDIS_SECURE}
       REDIS_USER: ${REDIS_USER}
-      REDIS_PASSWORD: ${REDIS_PASSWORD}
-      DIRECTOR_V2_HOST: ${DIRECTOR_V2_HOST}
-      DIRECTOR_V2_PORT: ${DIRECTOR_V2_PORT}
-      DYNAMIC_SCHEDULER_USE_INTERNAL_SCHEDULER: ${DYNAMIC_SCHEDULER_USE_INTERNAL_SCHEDULER}
-      DYNAMIC_SCHEDULER_LOGLEVEL: ${DYNAMIC_SCHEDULER_LOGLEVEL}
-      DYNAMIC_SCHEDULER_STOP_SERVICE_TIMEOUT: ${DYNAMIC_SCHEDULER_STOP_SERVICE_TIMEOUT}
-      DYNAMIC_SCHEDULER_PROFILING: ${DYNAMIC_SCHEDULER_PROFILING}
-      DYNAMIC_SCHEDULER_TRACING: ${DYNAMIC_SCHEDULER_TRACING}
-      DYNAMIC_SCHEDULER_UI_STORAGE_SECRET: ${DYNAMIC_SCHEDULER_UI_STORAGE_SECRET}
-      DYNAMIC_SIDECAR_API_SAVE_RESTORE_STATE_TIMEOUT: ${DYNAMIC_SIDECAR_API_SAVE_RESTORE_STATE_TIMEOUT}
+
       TRACING_OPENTELEMETRY_COLLECTOR_ENDPOINT: ${TRACING_OPENTELEMETRY_COLLECTOR_ENDPOINT}
       TRACING_OPENTELEMETRY_COLLECTOR_PORT: ${TRACING_OPENTELEMETRY_COLLECTOR_PORT}
   static-webserver:

services/dynamic-scheduler/requirements/_test.in

@@ -10,7 +10,7 @@
 #
 --constraint _base.txt
 
-
+aiopg[sa]
 asgi_lifespan
 coverage
 docker

services/dynamic-scheduler/requirements/_test.txt

@@ -1,9 +1,13 @@
+aiopg==1.4.0
+    # via -r requirements/_test.in
 anyio==4.6.2.post1
     # via
     #   -c requirements/_base.txt
     #   httpx
 asgi-lifespan==2.1.0
     # via -r requirements/_test.in
+async-timeout==4.0.3
+    # via aiopg
 certifi==2024.8.30
     # via
     #   -c requirements/../../../requirements/constraints.txt
@@ -27,6 +31,7 @@ greenlet==3.1.1
     # via
     #   -c requirements/_base.txt
     #   playwright
+    #   sqlalchemy
 h11==0.14.0
     # via
     #   -c requirements/_base.txt
@@ -73,6 +78,11 @@ pprintpp==0.4.0
     # via pytest-icdiff
 priority==2.0.0
     # via hypercorn
+psycopg2-binary==2.9.10
+    # via
+    #   -c requirements/_base.txt
+    #   aiopg
+    #   sqlalchemy
 pyee==12.0.0
     # via playwright
 pytest==8.3.4
@@ -121,6 +131,11 @@ sniffio==1.3.1
     #   anyio
     #   asgi-lifespan
     #   httpx
+sqlalchemy==1.4.54
+    # via
+    #   -c requirements/../../../requirements/constraints.txt
+    #   -c requirements/_base.txt
+    #   aiopg
 termcolor==2.5.0
     # via pytest-sugar
 typing-extensions==4.12.2

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

@@ -2,6 +2,7 @@
 import os
 
 import typer
+from settings_library.postgres import PostgresSettings
 from settings_library.rabbit import RabbitSettings
 from settings_library.utils_cli import (
     create_settings_command,
@@ -56,6 +57,21 @@ def echo_dotenv(ctx: typer.Context, *, minimal: bool = True):
             "DYNAMIC_SCHEDULER_UI_STORAGE_SECRET",
             "replace-with-ui-storage-secret",
         ),
+        DYNAMIC_SCHEDULER_POSTGRES=os.environ.get(
+            "DYNAMIC_SCHEDULER_POSTGRES",
+            PostgresSettings.create_from_envs(
+                POSTGRES_HOST=os.environ.get(
+                    "POSTGRES_HOST", "replace-with-postgres-host"
+                ),
+                POSTGRES_USER=os.environ.get(
+                    "POSTGRES_USER", "replace-with-postgres-user"
+                ),
+                POSTGRES_PASSWORD=os.environ.get(
+                    "POSTGRES_PASSWORD", "replace-with-postgres-password"
+                ),
+                POSTGRES_DB=os.environ.get("POSTGRES_DB", "replace-with-postgres-db"),
+            ),
+        ),
     )
 
     print_as_envfile(

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

@@ -27,6 +27,7 @@
 from ..services.director_v0 import lifespan_director_v0
 from ..services.director_v2 import lifespan_director_v2
 from ..services.notifier import get_notifier_lifespans
+from ..services.postgres import lifespan_postgres
 from ..services.rabbitmq import lifespan_rabbitmq
 from ..services.redis import lifespan_redis
 from ..services.service_tracker import lifespan_service_tracker
@@ -50,6 +51,7 @@ def create_app(settings: ApplicationSettings | None = None) -> FastAPI:
         lifespan_rabbitmq,
         lifespan_rpc_api_routes,
         lifespan_redis,
+        lifespan_postgres,
         *get_notifier_lifespans(),
         lifespan_service_tracker,
         lifespan_deferred_manager,

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

@@ -8,6 +8,7 @@
 from settings_library.director_v0 import DirectorV0Settings
 from settings_library.director_v2 import DirectorV2Settings
 from settings_library.http_client_request import ClientRequestSettings
+from settings_library.postgres import PostgresSettings
 from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
 from settings_library.tracing import TracingSettings
@@ -144,6 +145,14 @@ class ApplicationSettings(_BaseApplicationSettings):
         description="settings for director-v2 service",
     )
 
+    DYNAMIC_SCHEDULER_POSTGRES: Annotated[
+        PostgresSettings,
+        Field(
+            json_schema_extra={"auto_default_from_env": True},
+            description="settings for postgres service",
+        ),
+    ]
+
     DYNAMIC_SCHEDULER_PROMETHEUS_INSTRUMENTATION_ENABLED: bool = True
 
     DYNAMIC_SCHEDULER_PROFILING: bool = False

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

@@ -0,0 +1,8 @@
+from ._project_networks import ProjectNetworkNotFoundError, ProjectNetworksRepo
+from ._setup import lifespan_postgres
+
+__all__: tuple[str, ...] = (
+    "lifespan_postgres",
+    "ProjectNetworkNotFoundError",
+    "ProjectNetworksRepo",
+)

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/postgres/_project_networks.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)

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/postgres/_setup.py

@@ -0,0 +1,21 @@
+from collections.abc import AsyncIterator
+
+from fastapi import FastAPI
+from fastapi_lifespan_manager import State
+from servicelib.db_async_engine import close_db_connection, connect_to_db
+from sqlalchemy.ext.asyncio import AsyncEngine
+
+from ...core.settings import ApplicationSettings
+
+
+async def lifespan_postgres(app: FastAPI) -> AsyncIterator[State]:
+    settings: ApplicationSettings = app.state.settings
+
+    await connect_to_db(app, settings.DYNAMIC_SCHEDULER_POSTGRES)
+    assert app.state.engine  # nosec
+    assert isinstance(app.state.engine, AsyncEngine)  # nosec
+
+    yield {}
+
+    assert app.state.engine  # nosec
+    await close_db_connection(app)

services/dynamic-scheduler/tests/conftest.py

@@ -25,6 +25,8 @@
     "pytest_simcore.docker_swarm",
     "pytest_simcore.environment_configs",
     "pytest_simcore.faker_projects_data",
+    "pytest_simcore.faker_users_data",
+    "pytest_simcore.postgres_service",
     "pytest_simcore.rabbit_service",
     "pytest_simcore.redis_service",
     "pytest_simcore.repository_paths",
@@ -115,6 +117,11 @@ def disable_status_monitor_lifespan(mocker: MockerFixture) -> None:
     mocker.patch(f"{_PATH_APPLICATION}.lifespan_status_monitor")
 
 
+@pytest.fixture
+def disable_postgres_lifespan(mocker: MockerFixture) -> None:
+    mocker.patch(f"{_PATH_APPLICATION}.lifespan_postgres")
+
+
 MAX_TIME_FOR_APP_TO_STARTUP: Final[float] = 10
 MAX_TIME_FOR_APP_TO_SHUTDOWN: Final[float] = 10
 

services/dynamic-scheduler/tests/unit/api_frontend/conftest.py

@@ -9,12 +9,15 @@
 from unittest.mock import AsyncMock
 
 import pytest
+import sqlalchemy as sa
 from fastapi import FastAPI, status
 from httpx import AsyncClient
 from hypercorn.asyncio import serve
 from hypercorn.config import Config
 from playwright.async_api import Page, async_playwright
 from pytest_mock import MockerFixture
+from pytest_simcore.helpers.monkeypatch_envs import setenvs_from_dict
+from pytest_simcore.helpers.postgres_tools import PostgresTestConfig
 from pytest_simcore.helpers.typing_env import EnvVarsDict
 from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
@@ -54,11 +57,20 @@ def mock_remove_tracked_service(mocker: MockerFixture) -> AsyncMock:
 @pytest.fixture
 def app_environment(
     app_environment: EnvVarsDict,
+    postgres_db: sa.engine.Engine,
+    postgres_host_config: PostgresTestConfig,
     disable_status_monitor_background_task: None,
     rabbit_service: RabbitSettings,
     redis_service: RedisSettings,
     remove_redis_data: None,
+    monkeypatch: pytest.MonkeyPatch,
 ) -> EnvVarsDict:
+    setenvs_from_dict(
+        monkeypatch,
+        {
+            "POSTGRES_CLIENT_NAME": "test_postgres_client",
+        },
+    )
     return app_environment
 
 

services/dynamic-scheduler/tests/unit/api_frontend/test_api_frontend_routes_index.py

@@ -32,6 +32,7 @@
 from tenacity import AsyncRetrying, stop_after_delay, wait_fixed
 
 pytest_simcore_core_services_selection = [
+    "postgres",
     "rabbit",
     "redis",
 ]

services/dynamic-scheduler/tests/unit/api_frontend/test_api_frontend_routes_service.py

@@ -28,6 +28,7 @@
 from tenacity import AsyncRetrying, stop_after_delay, wait_fixed
 
 pytest_simcore_core_services_selection = [
+    "postgres",
     "rabbit",
     "redis",
 ]

services/dynamic-scheduler/tests/unit/api_rest/conftest.py

@@ -11,6 +11,7 @@
 
 @pytest.fixture
 def app_environment(
+    disable_postgres_lifespan: None,
     disable_rabbitmq_lifespan: None,
     disable_redis_lifespan: None,
     disable_service_tracker_lifespan: None,

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

@@ -179,6 +179,7 @@ def app_environment(
 
 @pytest.fixture
 async def rpc_client(
+    disable_postgres_lifespan: None,
     app_environment: EnvVarsDict,
     mock_director_v2_service_state: None,
     mock_director_v0_service_state: None,