Merge branch 'feature/store-socket-events' of github.com:odeimaiz/osparc-simcore into feature/store-socket-events

Author: odeimaiz

services/director-v2/src/simcore_service_director_v2/modules/db/__init__.py

@@ -1,15 +1,25 @@
 from fastapi import FastAPI
 from settings_library.postgres import PostgresSettings
 
+from ._asyncpg import (
+    asyncpg_close_db_connection,
+    asyncpg_connect_to_db,
+    get_asyncpg_engine,
+)
 from .events import close_db_connection, connect_to_db
 
 
 def setup(app: FastAPI, settings: PostgresSettings) -> None:
     async def on_startup() -> None:
         await connect_to_db(app, settings)
+        await asyncpg_connect_to_db(app, settings)
 
     async def on_shutdown() -> None:
+        await asyncpg_close_db_connection(app)
         await close_db_connection(app)
 
     app.add_event_handler("startup", on_startup)
     app.add_event_handler("shutdown", on_shutdown)
+
+
+__all__: tuple[str, ...] = ("get_asyncpg_engine",)

services/director-v2/src/simcore_service_director_v2/modules/db/_asyncpg.py

@@ -0,0 +1,36 @@
+import logging
+
+from fastapi import FastAPI
+from servicelib.db_asyncpg_utils import create_async_engine_and_pg_database_ready
+from servicelib.logging_utils import log_context
+from settings_library.postgres import PostgresSettings
+from simcore_postgres_database.utils_aiosqlalchemy import get_pg_engine_stateinfo
+
+_logger = logging.getLogger(__name__)
+
+
+async def asyncpg_connect_to_db(app: FastAPI, settings: PostgresSettings) -> None:
+    with log_context(
+        _logger,
+        logging.DEBUG,
+        f"Connecting and migraging {settings.dsn_with_async_sqlalchemy}",
+    ):
+        engine = await create_async_engine_and_pg_database_ready(settings)
+
+    app.state.asyncpg_engine = engine
+    _logger.debug(
+        "Setup engine: %s",
+        await get_pg_engine_stateinfo(engine),
+    )
+
+
+async def asyncpg_close_db_connection(app: FastAPI) -> None:
+    with log_context(
+        _logger, logging.DEBUG, f"db disconnect of {app.state.asyncpg_engine}"
+    ):
+        if engine := app.state.asyncpg_engine:
+            await engine.dispose()
+
+
+def get_asyncpg_engine(app: FastAPI):
+    return app.state.asyncpg_engine

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

@@ -9,7 +9,7 @@
 import sqlalchemy as sa
 from aiopg.sa.connection import SAConnection
 from aiopg.sa.result import RowProxy
-from models_library.projects import ProjectAtDB
+from models_library.projects import ProjectAtDB, ProjectID
 from models_library.projects_nodes import Node
 from models_library.projects_nodes_io import NodeIDStr
 from models_library.users import UserID
@@ -164,14 +164,18 @@ async def _get_tags_by_project(conn: SAConnection, project_id: str) -> list:
 
     @staticmethod
     async def _upsert_tags_in_project(
-        conn: SAConnection, project_index_id: int, project_tags: list[int]
+        conn: SAConnection,
+        project_index_id: int,
+        project_uuid: ProjectID,
+        project_tags: list[int],
     ) -> None:
         for tag_id in project_tags:
             await conn.execute(
                 pg_insert(projects_tags)
                 .values(
                     project_id=project_index_id,
                     tag_id=tag_id,
+                    project_uuid_for_rut=project_uuid,
                 )
                 .on_conflict_do_nothing()
             )

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

@@ -202,6 +202,7 @@ def _reraise_if_not_unique_uuid_error(err: UniqueViolation):
                         await self._upsert_tags_in_project(
                             conn=conn,
                             project_index_id=project_index,
+                            project_uuid=project_uuid,
                             project_tags=project_tag_ids,
                         )
                         selected_values["tags"] = project_tag_ids