🎨 catalog: lifespan managers for fastapi apps (#7483)

Author: pcrespov

packages/pytest-simcore/src/pytest_simcore/helpers/logging_tools.py

@@ -22,7 +22,7 @@ def _timedelta_as_minute_second_ms(delta: datetime.timedelta) -> str:
     if int(milliseconds * 1000) != 0:
         result += f"{int(milliseconds*1000)}ms"
 
-    sign = "-" if total_seconds < 0 else ""
+    sign = "-" if total_seconds < 0 else "<1ms"
 
     return f"{sign}{result.strip()}"
 
@@ -32,10 +32,11 @@ class DynamicIndentFormatter(logging.Formatter):
     _cls_indent_level: int = 0
     _instance_indent_level: int = 0
 
-    def __init__(self, fmt=None, datefmt=None, style="%"):
+    def __init__(self, *args, **kwargs):
+        fmt = args[0] if args else None
         dynamic_fmt = fmt or "%(asctime)s %(levelname)s %(message)s"
         assert "message" in dynamic_fmt
-        super().__init__(dynamic_fmt, datefmt, style)
+        super().__init__(dynamic_fmt, *args, **kwargs)
 
     def format(self, record) -> str:
         original_message = record.msg

packages/service-library/src/servicelib/aiohttp/db_asyncpg_engine.py

@@ -4,7 +4,6 @@
 SEE migration aiopg->asyncpg https://github.com/ITISFoundation/osparc-simcore/issues/4529
 """
 
-
 import logging
 from typing import Final
 
@@ -16,7 +15,7 @@
 )
 from sqlalchemy.ext.asyncio import AsyncEngine
 
-from ..db_asyncpg_utils import create_async_engine_and_pg_database_ready
+from ..db_asyncpg_utils import create_async_engine_and_database_ready
 from ..logging_utils import log_context
 
 APP_DB_ASYNC_ENGINE_KEY: Final[str] = f"{__name__ }.AsyncEngine"
@@ -56,7 +55,7 @@ async def connect_to_db(app: web.Application, settings: PostgresSettings) -> Non
         "Connecting app[APP_DB_ASYNC_ENGINE_KEY] to postgres with %s",
         f"{settings=}",
     ):
-        engine = await create_async_engine_and_pg_database_ready(settings)
+        engine = await create_async_engine_and_database_ready(settings)
         _set_async_engine_to_app_state(app, engine)
 
     _logger.info(

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

@@ -1,4 +1,5 @@
 import logging
+import warnings
 
 from fastapi import FastAPI
 from settings_library.postgres import PostgresSettings
@@ -17,6 +18,12 @@
 
 @retry(**PostgresRetryPolicyUponInitialization(_logger).kwargs)
 async def connect_to_db(app: FastAPI, settings: PostgresSettings) -> None:
+    warnings.warn(
+        "The 'connect_to_db' function is deprecated and will be removed in a future release. "
+        "Please use 'postgres_lifespan' instead for managing the database connection lifecycle.",
+        DeprecationWarning,
+        stacklevel=2,
+    )
     with log_context(
         _logger, logging.DEBUG, f"connection to db {settings.dsn_with_async_sqlalchemy}"
     ):

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

@@ -4,9 +4,6 @@
 
 from models_library.healthchecks import IsNonResponsive, IsResponsive, LivenessResult
 from settings_library.postgres import PostgresSettings
-from simcore_postgres_database.utils_aiosqlalchemy import (  # type: ignore[import-not-found] # this on is unclear
-    raise_if_migration_not_ready,
-)
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
 from tenacity import retry
@@ -17,7 +14,7 @@
 
 
 @retry(**PostgresRetryPolicyUponInitialization(_logger).kwargs)
-async def create_async_engine_and_pg_database_ready(
+async def create_async_engine_and_database_ready(
     settings: PostgresSettings,
 ) -> AsyncEngine:
     """
@@ -26,13 +23,17 @@ async def create_async_engine_and_pg_database_ready(
     - waits until db data is migrated (i.e. ready to use)
     - returns engine
     """
+    from simcore_postgres_database.utils_aiosqlalchemy import (  # type: ignore[import-not-found] # this on is unclear
+        raise_if_migration_not_ready,
+    )
+
     server_settings = None
     if settings.POSTGRES_CLIENT_NAME:
         server_settings = {
             "application_name": settings.POSTGRES_CLIENT_NAME,
         }
 
-    engine: AsyncEngine = create_async_engine(
+    engine = create_async_engine(
         settings.dsn_with_async_sqlalchemy,
         pool_size=settings.POSTGRES_MINSIZE,
         max_overflow=settings.POSTGRES_MAXSIZE - settings.POSTGRES_MINSIZE,
@@ -43,9 +44,10 @@ async def create_async_engine_and_pg_database_ready(
 
     try:
         await raise_if_migration_not_ready(engine)
-    except Exception:
+    except Exception as exc:
         # NOTE: engine must be closed because retry will create a new engine
         await engine.dispose()
+        exc.add_note("Failed during migration check. Created engine disposed.")
         raise
 
     return engine

packages/service-library/src/servicelib/fastapi/db_asyncpg_engine.py

@@ -1,4 +1,5 @@
 import logging
+import warnings
 
 from fastapi import FastAPI
 from settings_library.postgres import PostgresSettings
@@ -7,19 +8,26 @@
 )
 from sqlalchemy.ext.asyncio import AsyncEngine
 
-from ..db_asyncpg_utils import create_async_engine_and_pg_database_ready
+from ..db_asyncpg_utils import create_async_engine_and_database_ready
 from ..logging_utils import log_context
 
 _logger = logging.getLogger(__name__)
 
 
 async def connect_to_db(app: FastAPI, settings: PostgresSettings) -> None:
+    warnings.warn(
+        "The 'connect_to_db' function is deprecated and will be removed in a future release. "
+        "Please use 'postgres_lifespan' instead for managing the database connection lifecycle.",
+        DeprecationWarning,
+        stacklevel=2,
+    )
+
     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)
+        engine = await create_async_engine_and_database_ready(settings)
 
     app.state.engine = engine
     _logger.debug(

packages/service-library/src/servicelib/fastapi/lifespan_utils.py

@@ -1,13 +1,24 @@
 from collections.abc import AsyncIterator
 from typing import Protocol
 
+from common_library.errors_classes import OsparcErrorMixin
 from fastapi import FastAPI
 from fastapi_lifespan_manager import LifespanManager, State
 
 
+class LifespanError(OsparcErrorMixin, RuntimeError): ...
+
+
+class LifespanOnStartupError(LifespanError):
+    msg_template = "Failed during startup of {module}"
+
+
+class LifespanOnShutdownError(LifespanError):
+    msg_template = "Failed during shutdown of {module}"
+
+
 class LifespanGenerator(Protocol):
-    def __call__(self, app: FastAPI) -> AsyncIterator["State"]:
-        ...
+    def __call__(self, app: FastAPI) -> AsyncIterator["State"]: ...
 
 
 def combine_lifespans(*generators: LifespanGenerator) -> LifespanManager:

packages/service-library/src/servicelib/fastapi/postgres_lifespan.py

@@ -0,0 +1,54 @@
+import asyncio
+import logging
+from collections.abc import AsyncIterator
+from enum import Enum
+
+from fastapi_lifespan_manager import State
+from servicelib.logging_utils import log_catch, log_context
+from settings_library.postgres import PostgresSettings
+from sqlalchemy.ext.asyncio import AsyncEngine
+
+from ..db_asyncpg_utils import create_async_engine_and_database_ready
+from .lifespan_utils import LifespanOnStartupError
+
+_logger = logging.getLogger(__name__)
+
+
+class PostgresLifespanState(str, Enum):
+    POSTGRES_SETTINGS = "postgres_settings"
+    POSTGRES_ASYNC_ENGINE = "postgres.async_engine"
+
+
+class PostgresConfigurationError(LifespanOnStartupError):
+    msg_template = "Invalid postgres settings [={settings}] on startup. Note that postgres cannot be disabled using settings"
+
+
+def create_input_state(settings: PostgresSettings) -> State:
+    return {PostgresLifespanState.POSTGRES_SETTINGS: settings}
+
+
+async def postgres_database_lifespan(_, state: State) -> AsyncIterator[State]:
+
+    with log_context(_logger, logging.INFO, f"{__name__}"):
+
+        settings = state[PostgresLifespanState.POSTGRES_SETTINGS]
+
+        if settings is None or not isinstance(settings, PostgresSettings):
+            raise PostgresConfigurationError(settings=settings)
+
+        assert isinstance(settings, PostgresSettings)  # nosec
+
+        # connect to database
+        async_engine: AsyncEngine = await create_async_engine_and_database_ready(
+            settings
+        )
+
+        try:
+
+            yield {
+                PostgresLifespanState.POSTGRES_ASYNC_ENGINE: async_engine,
+            }
+
+        finally:
+            with log_catch(_logger, reraise=False):
+                await asyncio.wait_for(async_engine.dispose(), timeout=10)