use api to get engine from app

Author: pcrespov

services/web/server/src/simcore_service_webserver/api_keys/_db.py

@@ -10,10 +10,11 @@
 from models_library.basic_types import IdInt
 from models_library.products import ProductName
 from models_library.users import UserID
-from servicelib.aiohttp.application_keys import APP_AIOPG_ENGINE_KEY
 from simcore_postgres_database.models.api_keys import api_keys
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 
+from ..db.plugin import get_database_engine
+
 _logger = logging.getLogger(__name__)
 
 
@@ -23,7 +24,7 @@ class ApiKeyRepo:
 
     @classmethod
     def create_from_app(cls, app: web.Application):
-        return cls(engine=app[APP_AIOPG_ENGINE_KEY])
+        return cls(engine=get_database_engine(app))
 
     async def list_names(
         self, *, user_id: UserID, product_name: ProductName

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

@@ -2,7 +2,8 @@
 from aiopg.sa.engine import Engine
 from models_library.users import UserID
 
-from .._constants import APP_AIOPG_ENGINE_KEY, RQT_USERID_KEY
+from .._constants import RQT_USERID_KEY
+from . import _aiopg
 
 
 class BaseRepository:
@@ -15,13 +16,13 @@ def __init__(self, engine: Engine, user_id: UserID | None = None):
     @classmethod
     def create_from_request(cls, request: web.Request):
         return cls(
-            engine=request.app[APP_AIOPG_ENGINE_KEY],
+            engine=_aiopg.get_database_engine(request.app),
             user_id=request.get(RQT_USERID_KEY),
         )
 
     @classmethod
     def create_from_app(cls, app: web.Application):
-        return cls(engine=app[APP_AIOPG_ENGINE_KEY], user_id=None)
+        return cls(engine=_aiopg.get_database_engine(app), user_id=None)
 
     @property
     def engine(self) -> Engine:

services/web/server/src/simcore_service_webserver/db_listener/_db_comp_tasks_listening_task.py

@@ -18,10 +18,10 @@
 from models_library.projects_nodes_io import NodeID
 from models_library.projects_state import RunningState
 from pydantic.types import PositiveInt
-from servicelib.aiohttp.application_keys import APP_AIOPG_ENGINE_KEY
 from simcore_postgres_database.webserver_models import DB_CHANNEL_NAME, projects
 from sqlalchemy.sql import select
 
+from ..db.plugin import get_database_engine
 from ..projects import exceptions, projects_api
 from ..projects.nodes_utils import update_node_outputs
 from ._utils import convert_state_from_db
@@ -159,7 +159,7 @@ async def _comp_tasks_listening_task(app: web.Application) -> None:
     while True:
         try:
             # create a special connection here
-            db_engine = app[APP_AIOPG_ENGINE_KEY]
+            db_engine = get_database_engine(app)
             _logger.info("listening to comp_task events...")
             await _listen(app, db_engine)
         except asyncio.CancelledError:  # noqa: PERF203

services/web/server/src/simcore_service_webserver/garbage_collector/_tasks_users.py

@@ -15,7 +15,7 @@
 from tenacity.before_sleep import before_sleep_log
 from tenacity.wait import wait_exponential
 
-from .._constants import APP_AIOPG_ENGINE_KEY
+from ..db.plugin import get_database_engine
 from ..login.utils import notify_user_logout
 from ..security.api import clean_auth_policy_cache
 from ..users.api import update_expired_users
@@ -60,7 +60,7 @@ async def _update_expired_users(app: web.Application):
     """
     It is resilient, i.e. if update goes wrong, it waits a bit and retries
     """
-    engine: Engine = app[APP_AIOPG_ENGINE_KEY]
+    engine: Engine = get_database_engine(app)
     assert engine  # nosec
 
     if updated := await update_expired_users(engine):

services/web/server/src/simcore_service_webserver/products/_events.py

@@ -2,7 +2,6 @@
 import tempfile
 from collections import OrderedDict
 from pathlib import Path
-from typing import cast
 
 from aiohttp import web
 from aiopg.sa.engine import Engine
@@ -14,7 +13,8 @@
     get_or_create_product_group,
 )
 
-from .._constants import APP_AIOPG_ENGINE_KEY, APP_PRODUCTS_KEY
+from .._constants import APP_PRODUCTS_KEY
+from ..db.plugin import get_database_engine
 from ..statics._constants import FRONTEND_APP_DEFAULT, FRONTEND_APPS_AVAILABLE
 from ._db import get_product_payment_fields, iter_products
 from ._model import Product
@@ -46,7 +46,7 @@ async def auto_create_products_groups(app: web.Application) -> None:
     NOTE: could not add this in 'setup_groups' (groups plugin)
     since it has to be executed BEFORE 'load_products_on_startup'
     """
-    engine = cast(Engine, app[APP_AIOPG_ENGINE_KEY])
+    engine = get_database_engine(app)
 
     async with engine.acquire() as connection:
         async for row in iter_products(connection):
@@ -76,7 +76,7 @@ async def load_products_on_startup(app: web.Application):
     Loads info on products stored in the database into app's storage (i.e. memory)
     """
     app_products: OrderedDict[str, Product] = OrderedDict()
-    engine: Engine = app[APP_AIOPG_ENGINE_KEY]
+    engine: Engine = get_database_engine(app)
     async with engine.acquire() as connection:
         async for row in iter_products(connection):
             assert isinstance(row, RowProxy)  # nosec

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

@@ -7,10 +7,10 @@
 import sqlalchemy as sa
 from aiohttp import web
 from aiopg.sa.result import ResultProxy, RowProxy
-from servicelib.aiohttp.application_keys import APP_AIOPG_ENGINE_KEY
 from simcore_postgres_database.models.scicrunch_resources import scicrunch_resources
 from sqlalchemy.dialects.postgresql import insert as sa_pg_insert
 
+from ..db.plugin import get_database_engine
 from .models import ResearchResource, ResearchResourceAtdB
 
 logger = logging.getLogger(__name__)
@@ -26,7 +26,7 @@ class ResearchResourceRepository:
     # WARNING: interfaces to both ResarchResource and ResearchResourceAtDB
 
     def __init__(self, app: web.Application):
-        self._engine = app[APP_AIOPG_ENGINE_KEY]
+        self._engine = get_database_engine(app)
 
     async def list_resources(self) -> list[ResearchResource]:
         async with self._engine.acquire() as conn: