fixed connection to db

Author: sanderegg

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

@@ -1,13 +1,15 @@
+import contextlib
 import logging
 import warnings
+from collections.abc import AsyncIterator
 
 from fastapi import FastAPI
 from settings_library.postgres import PostgresSettings
 from simcore_postgres_database.utils_aiosqlalchemy import (  # type: ignore[import-not-found] # this on is unclear
     get_pg_engine_stateinfo,
     raise_if_migration_not_ready,
 )
-from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
 from tenacity import retry
 
 from .logging_utils import log_context
@@ -16,6 +18,37 @@
 _logger = logging.getLogger(__name__)
 
 
+@contextlib.asynccontextmanager
+async def with_async_pg_engine(
+    settings: PostgresSettings,
+) -> AsyncIterator[AsyncEngine]:
+    """
+    Creates an asyncpg engine and ensures it is properly closed after use.
+    """
+    try:
+        with log_context(
+            _logger,
+            logging.DEBUG,
+            f"connection to db {settings.dsn_with_async_sqlalchemy}",
+        ):
+            server_settings = None
+            if settings.POSTGRES_CLIENT_NAME:
+                assert isinstance(settings.POSTGRES_CLIENT_NAME, str)
+
+            engine = create_async_engine(
+                settings.dsn_with_async_sqlalchemy,
+                pool_size=settings.POSTGRES_MINSIZE,
+                max_overflow=settings.POSTGRES_MAXSIZE - settings.POSTGRES_MINSIZE,
+                connect_args={"server_settings": server_settings},
+                pool_pre_ping=True,  # https://docs.sqlalchemy.org/en/14/core/pooling.html#dealing-with-disconnects
+                future=True,  # this uses sqlalchemy 2.0 API, shall be removed when sqlalchemy 2.0 is released
+            )
+        yield engine
+    finally:
+        with log_context(_logger, logging.DEBUG, f"db disconnect of {engine}"):
+            await engine.dispose()
+
+
 @retry(**PostgresRetryPolicyUponInitialization(_logger).kwargs)
 async def connect_to_db(app: FastAPI, settings: PostgresSettings) -> None:
     warnings.warn(

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/core/external_dependencies.py

@@ -1,13 +1,10 @@
 from common_library.errors_classes import OsparcErrorMixin
 from fastapi import FastAPI
-from servicelib.db_async_engine import connect_to_db
 from servicelib.utils import logged_gather
-from settings_library.postgres import PostgresSettings
 
-from ..modules.service_liveness import wait_for_service_liveness
+from ..modules.database import wait_for_database_liveness
 from .rabbitmq import wait_for_rabbitmq_liveness
 from .registry import wait_for_registries_liveness
-from .settings import ApplicationSettings
 from .storage import wait_for_storage_liveness
 
 
@@ -24,19 +21,9 @@ def setup_check_dependencies(app: FastAPI) -> None:
     # start rapidly, for this reason they are run in
     # parallel.
     async def on_startup() -> None:
-        app_settings = app.state.settings
-        assert isinstance(app_settings, ApplicationSettings)  # nosec
-        postgres_settings = app_settings.POSTGRES_SETTINGS
-        assert isinstance(postgres_settings, PostgresSettings)  # nosec
         liveliness_results = await logged_gather(
             *[
-                wait_for_service_liveness(
-                    connect_to_db,
-                    app,
-                    postgres_settings,
-                    service_name="Postgres",
-                    endpoint=postgres_settings.dsn,
-                ),
+                wait_for_database_liveness(app),
                 wait_for_rabbitmq_liveness(app),
                 wait_for_registries_liveness(app),
                 wait_for_storage_liveness(app),

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/database.py

@@ -0,0 +1,27 @@
+from fastapi import FastAPI
+from servicelib.db_async_engine import with_async_pg_engine
+from servicelib.db_asyncpg_utils import check_postgres_liveness
+from settings_library.postgres import PostgresSettings
+
+from ..core.settings import ApplicationSettings
+from .service_liveness import (
+    wait_for_service_liveness,
+)
+
+
+async def wait_for_database_liveness(app: FastAPI) -> None:
+    """
+    Checks if the postgres engine is alive and can be used.
+    """
+
+    app_settings = app.state.settings
+    assert isinstance(app_settings, ApplicationSettings)  # nosec
+    postgres_settings = app_settings.POSTGRES_SETTINGS
+    assert isinstance(postgres_settings, PostgresSettings)  # nosec
+    async with with_async_pg_engine(postgres_settings) as engine:
+        await wait_for_service_liveness(
+            check_postgres_liveness,
+            engine,
+            service_name="Postgres",
+            endpoint=postgres_settings.dsn,
+        )