Merge branch 'master' into 2024/fix/tracing

Author: mrnicegyu11

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