♻️🗑️ api-server upgraded to use asyncpg (#7598)

Author: pcrespov

services/api-server/requirements/_base.in

@@ -17,7 +17,6 @@
 --requirement ../../../packages/service-library/requirements/_fastapi.in
 
 aiofiles
-aiopg[sa]
 cryptography
 fastapi-pagination
 fastapi[all]

services/api-server/requirements/_base.txt

@@ -55,8 +55,6 @@ aiohttp==3.11.10
     #   -c requirements/../../../requirements/constraints.txt
     #   -r requirements/../../../packages/simcore-sdk/requirements/_base.in
     #   aiodocker
-aiopg==1.4.0
-    # via -r requirements/_base.in
 aiormq==6.8.1
     # via aio-pika
 aiosignal==1.3.1
@@ -84,8 +82,6 @@ arrow==1.3.0
     #   -r requirements/../../../packages/simcore-sdk/requirements/../../../packages/service-library/requirements/_base.in
 asgiref==3.8.1
     # via opentelemetry-instrumentation-asgi
-async-timeout==4.0.3
-    # via aiopg
 asyncpg==0.30.0
     # via sqlalchemy
 attrs==24.2.0
@@ -513,9 +509,7 @@ psutil==6.1.0
     #   -r requirements/../../../packages/service-library/requirements/_base.in
     #   -r requirements/../../../packages/simcore-sdk/requirements/../../../packages/service-library/requirements/_base.in
 psycopg2-binary==2.9.10
-    # via
-    #   aiopg
-    #   sqlalchemy
+    # via sqlalchemy
 pycparser==2.22
     # via cffi
 pycryptodome==3.21.0
@@ -785,7 +779,6 @@ sqlalchemy==1.4.54
     #   -r requirements/../../../packages/postgres-database/requirements/_base.in
     #   -r requirements/../../../packages/simcore-sdk/requirements/../../../packages/postgres-database/requirements/_base.in
     #   -r requirements/../../../packages/simcore-sdk/requirements/_base.in
-    #   aiopg
     #   alembic
 starlette==0.41.3
     # via

services/api-server/src/simcore_service_api_server/api/dependencies/authentication.py

@@ -6,8 +6,8 @@
 from models_library.products import ProductName
 from pydantic import BaseModel, PositiveInt
 
-from ...db.repositories.api_keys import ApiKeysRepository, UserAndProductTuple
-from ...db.repositories.users import UsersRepository
+from ...repository.api_keys import ApiKeysRepository, UserAndProductTuple
+from ...repository.users import UsersRepository
 from .database import get_repository
 
 # SEE https://swagger.io/docs/specification/authentication/basic-authentication/
@@ -22,9 +22,9 @@ class Identity(BaseModel):
 
 def _create_exception() -> HTTPException:
     _unauthorized_headers = {
-        "WWW-Authenticate": f'Basic realm="{basic_scheme.realm}"'
-        if basic_scheme.realm
-        else "Basic"
+        "WWW-Authenticate": (
+            f'Basic realm="{basic_scheme.realm}"' if basic_scheme.realm else "Basic"
+        )
     }
     return HTTPException(
         status_code=status.HTTP_401_UNAUTHORIZED,
@@ -46,7 +46,7 @@ async def get_current_identity(
     if user_and_product is None:
         exc = _create_exception()
         raise exc
-    email = await users_repo.get_active_user_email(user_and_product.user_id)
+    email = await users_repo.get_active_user_email(user_id=user_and_product.user_id)
     if not email:
         exc = _create_exception()
         raise exc

services/api-server/src/simcore_service_api_server/api/dependencies/database.py

@@ -1,53 +1,38 @@
 import logging
 from collections.abc import AsyncGenerator, Callable
-from typing import Annotated, cast
+from typing import Annotated
 
-from aiopg.sa import Engine
 from fastapi import Depends
 from fastapi.requests import Request
+from simcore_postgres_database.utils_aiosqlalchemy import (
+    get_pg_engine_stateinfo,
+)
 from sqlalchemy.ext.asyncio import AsyncEngine
 
-from ...db.events import get_asyncpg_engine
-from ...db.repositories import BaseRepository
-
-logger = logging.getLogger(__name__)
-
+from ...clients.postgres import get_engine
+from ...repository import BaseRepository
 
-def get_db_engine(request: Request) -> Engine:
-    return cast(Engine, request.app.state.engine)
+_logger = logging.getLogger(__name__)
 
 
 def get_db_asyncpg_engine(request: Request) -> AsyncEngine:
-    return get_asyncpg_engine(request.app)
+    return get_engine(request.app)
 
 
 def get_repository(repo_type: type[BaseRepository]) -> Callable:
     async def _get_repo(
-        engine: Annotated[Engine, Depends(get_db_engine)],
+        engine: Annotated[AsyncEngine, Depends(get_db_asyncpg_engine)],
     ) -> AsyncGenerator[BaseRepository, None]:
         # NOTE: 2 different ideas were tried here with not so good
         # 1st one was acquiring a connection per repository which lead to the following issue https://github.com/ITISFoundation/osparc-simcore/pull/1966
         # 2nd one was acquiring a connection per request which works but blocks the director-v2 responsiveness once
         # the max amount of connections is reached
         # now the current solution is to acquire connection when needed.
+        _logger.debug(
+            "Setting up a repository. Current state of connections: %s",
+            await get_pg_engine_stateinfo(engine),
+        )
 
-        available_engines = engine.maxsize - (engine.size - engine.freesize)
-        if available_engines <= 1:
-            logger.warning(
-                "Low pg connections available in pool: pool size=%d, acquired=%d, free=%d, reserved=[%d, %d]",
-                engine.size,
-                engine.size - engine.freesize,
-                engine.freesize,
-                engine.minsize,
-                engine.maxsize,
-            )
         yield repo_type(db_engine=engine)
 
     return _get_repo
-
-
-__all__: tuple[str, ...] = (
-    "Engine",
-    "get_db_engine",
-    "get_repository",
-)

services/api-server/src/simcore_service_api_server/db/__init__.py renamed to services/api-server/src/simcore_service_api_server/clients/__init__.py

@@ -0,0 +1,42 @@
+from fastapi import FastAPI
+from servicelib.fastapi.db_asyncpg_engine import close_db_connection, connect_to_db
+from servicelib.fastapi.lifespan_utils import LifespanOnStartupError
+from sqlalchemy.ext.asyncio import AsyncEngine
+
+from ..core.settings import ApplicationSettings
+
+
+class PostgresNotConfiguredError(LifespanOnStartupError):
+    msg_template = LifespanOnStartupError.msg_template + (
+        "Postgres settings are not configured. "
+        "Please check your application settings. "
+    )
+
+
+def get_engine(app: FastAPI) -> AsyncEngine:
+    assert app.state.engine  # nosec
+    engine: AsyncEngine = app.state.engine
+    return engine
+
+
+def setup_postgres(app: FastAPI):
+    app.state.engine = None
+
+    async def _on_startup() -> None:
+        settings: ApplicationSettings = app.state.settings
+        if settings.API_SERVER_POSTGRES is None:
+            raise PostgresNotConfiguredError(
+                lifespan_name="Postgres",
+                settings=settings,
+            )
+
+        await connect_to_db(app, settings.API_SERVER_POSTGRES)
+        assert app.state.engine  # nosec
+        assert isinstance(app.state.engine, AsyncEngine)  # nosec
+
+    async def _on_shutdown() -> None:
+        assert app.state.engine  # nosec
+        await close_db_connection(app)
+
+    app.add_event_handler("startup", _on_startup)
+    app.add_event_handler("shutdown", _on_shutdown)

services/api-server/src/simcore_service_api_server/clients/postgres.py

@@ -7,6 +7,7 @@
 from servicelib.fastapi.profiler import initialize_profiler
 from servicelib.fastapi.tracing import initialize_tracing
 from servicelib.logging_utils import config_all_loggers
+from simcore_service_api_server.clients.postgres import setup_postgres
 
 from .. import exceptions
 from .._meta import API_VERSION, API_VTAG, APP_NAME
@@ -15,7 +16,7 @@
 from ..services_http import catalog, director_v2, storage, webserver
 from ..services_http.rabbitmq import setup_rabbitmq
 from ._prometheus_instrumentation import setup_prometheus_instrumentation
-from .events import create_start_app_handler, create_stop_app_handler
+from .events import on_shutdown, on_startup
 from .openapi import override_openapi_method, use_route_names_as_operation_ids
 from .settings import ApplicationSettings
 
@@ -81,6 +82,9 @@ def init_app(settings: ApplicationSettings | None = None) -> FastAPI:
 
     app.state.settings = settings
 
+    if settings.API_SERVER_POSTGRES:
+        setup_postgres(app)
+
     setup_rabbitmq(app)
 
     if settings.API_SERVER_TRACING:
@@ -115,8 +119,8 @@ def init_app(settings: ApplicationSettings | None = None) -> FastAPI:
         )
 
     # setup app
-    app.add_event_handler("startup", create_start_app_handler(app))
-    app.add_event_handler("shutdown", create_stop_app_handler(app))
+    app.add_event_handler("startup", on_startup)
+    app.add_event_handler("shutdown", on_shutdown)
 
     exceptions.setup_exception_handlers(
         app, is_debug=settings.SC_BOOT_MODE == BootModeEnum.DEBUG

services/api-server/src/simcore_service_api_server/core/application.py

@@ -1,54 +1,15 @@
 import logging
-from collections.abc import Callable
-
-from fastapi import FastAPI
 
 from .._meta import APP_FINISHED_BANNER_MSG, APP_STARTED_BANNER_MSG
-from ..db.events import (
-    asyncpg_close_db_connection,
-    asyncpg_connect_to_db,
-    close_db_connection,
-    connect_to_db,
-)
-from .settings import ApplicationSettings
-
-logger = logging.getLogger(__name__)
-
-
-def create_start_app_handler(app: FastAPI) -> Callable:
-    async def _on_startup() -> None:
-        logger.info("Application starting ...")
-        if app.state.settings.API_SERVER_POSTGRES:
-            # database
-            assert isinstance(app.state.settings, ApplicationSettings)  # nosec
-            await connect_to_db(app)
-            await asyncpg_connect_to_db(app, app.state.settings.API_SERVER_POSTGRES)
-            assert app.state.engine  # nosec
-            assert app.state.asyncpg_engine  # nosec
-
-        print(APP_STARTED_BANNER_MSG, flush=True)  # noqa: T201
-
-    return _on_startup
-
 
-def create_stop_app_handler(app: FastAPI) -> Callable:
-    async def _on_shutdown() -> None:
-        logger.info("Application stopping, ...")
+_logger = logging.getLogger(__name__)
 
-        if app.state.settings.API_SERVER_POSTGRES:
-            assert isinstance(app.state.settings, ApplicationSettings)  # nosec
-            try:
-                await asyncpg_close_db_connection(app)
-                await close_db_connection(app)
 
-            except Exception as err:  # pylint: disable=broad-except
-                logger.warning(
-                    "Failed to close app: %s",
-                    err,
-                    exc_info=app.state.settings.debug,
-                    stack_info=app.state.settings.debug,
-                )
+async def on_startup() -> None:
+    _logger.info("Application starting ...")
+    print(APP_STARTED_BANNER_MSG, flush=True)  # noqa: T201
 
-        print(APP_FINISHED_BANNER_MSG, flush=True)  # noqa: T201
 
-    return _on_shutdown
+async def on_shutdown() -> None:
+    _logger.info("Application stopping, ...")
+    print(APP_FINISHED_BANNER_MSG, flush=True)  # noqa: T201