aiopg vs asyncpg

Author: pcrespov

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

@@ -5,7 +5,7 @@
 from simcore_postgres_database.models.workspaces_access_rights import (
     workspaces_access_rights,
 )
-from simcore_service_webserver.db.plugin import get_database_engine
+from simcore_service_webserver.db.plugin import get_aiopg_engine
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 
 
@@ -18,7 +18,7 @@ async def update_or_insert_workspace_group(
     write: bool,
     delete: bool,
 ) -> None:
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         insert_stmt = pg_insert(workspaces_access_rights).values(
             workspace_id=workspace_id,
             gid=group_id,

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

@@ -30,10 +30,10 @@ def setup_db(app: web.Application):
 
 
 # aiopg helpers
-get_database_engine = _aiopg.get_database_engine
-get_engine_state = _aiopg.get_engine_state
+get_aiopg_engine = _aiopg.get_database_engine
+get_aiopg_engine_state = _aiopg.get_engine_state
 is_service_responsive = _aiopg.is_service_responsive
 is_service_enabled = _aiopg.is_service_enabled
 
 # asyncpg helpers
-get_async_engine = _asyncpg.get_async_engine
+get_asyncpg_engine = _asyncpg.get_async_engine

services/web/server/src/simcore_service_webserver/diagnostics/_handlers.py

@@ -116,7 +116,7 @@ def _get_client_session_info():
     async def _check_pg():
         check.services["postgres"] = {
             "healthy": await plugin.is_service_responsive(request.app),
-            "pool": plugin.get_engine_state(request.app),
+            "pool": plugin.get_aiopg_engine_state(request.app),
         }
 
     async def _check_storage():

services/web/server/src/simcore_service_webserver/director_v2/_handlers.py

@@ -27,7 +27,7 @@
 
 from .._constants import RQ_PRODUCT_KEY
 from .._meta import API_VTAG as VTAG
-from ..db.plugin import get_database_engine
+from ..db.plugin import get_aiopg_engine
 from ..login.decorators import login_required
 from ..products import api as products_api
 from ..security.decorators import permission_required
@@ -91,7 +91,7 @@ async def start_computation(request: web.Request) -> web.Response:
             X_SIMCORE_USER_AGENT, UNDEFINED_DEFAULT_SIMCORE_USER_AGENT_VALUE
         )
 
-        async with get_database_engine(request.app).acquire() as conn:
+        async with get_aiopg_engine(request.app).acquire() as conn:
             group_properties = (
                 await GroupExtraPropertiesRepo.get_aggregated_properties_for_user(
                     conn, user_id=req_ctx.user_id, product_name=req_ctx.product_name

services/web/server/src/simcore_service_webserver/folders/_folders_db.py

@@ -22,7 +22,7 @@
 from sqlalchemy.orm import aliased
 from sqlalchemy.sql import asc, desc, select
 
-from ..db.plugin import get_database_engine
+from ..db.plugin import get_aiopg_engine
 from .errors import FolderAccessForbiddenError, FolderNotFoundError
 
 _logger = logging.getLogger(__name__)
@@ -54,7 +54,7 @@ async def create(
         user_id is not None and workspace_id is not None
     ), "Both user_id and workspace_id cannot be provided at the same time. Please provide only one."
 
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         result = await conn.execute(
             folders_v2.insert()
             .values(
@@ -117,7 +117,7 @@ async def list_(
         list_query = base_query.order_by(desc(getattr(folders_v2.c, order_by.field)))
     list_query = list_query.offset(offset).limit(limit)
 
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         count_result = await conn.execute(count_query)
         total_count = await count_result.scalar()
 
@@ -142,7 +142,7 @@ async def get(
         )
     )
 
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         result = await conn.execute(query)
         row = await result.first()
         if row is None:
@@ -178,7 +178,7 @@ async def get_for_user_or_workspace(
     else:
         query = query.where(folders_v2.c.workspace_id == workspace_id)
 
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         result = await conn.execute(query)
         row = await result.first()
         if row is None:
@@ -196,7 +196,7 @@ async def update(
     parent_folder_id: FolderID | None,
     product_name: ProductName,
 ) -> FolderDB:
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         result = await conn.execute(
             folders_v2.update()
             .values(
@@ -222,7 +222,7 @@ async def delete_recursively(
     folder_id: FolderID,
     product_name: ProductName,
 ) -> None:
-    async with get_database_engine(app).acquire() as conn, conn.begin():
+    async with get_aiopg_engine(app).acquire() as conn, conn.begin():
         # Step 1: Define the base case for the recursive CTE
         base_query = select(
             folders_v2.c.folder_id, folders_v2.c.parent_folder_id
@@ -276,7 +276,7 @@ async def get_projects_recursively_only_if_user_is_owner(
     or the `users_to_groups` table for private workspace projects.
     """
 
-    async with get_database_engine(app).acquire() as conn, conn.begin():
+    async with get_aiopg_engine(app).acquire() as conn, conn.begin():
         # Step 1: Define the base case for the recursive CTE
         base_query = select(
             folders_v2.c.folder_id, folders_v2.c.parent_folder_id

services/web/server/src/simcore_service_webserver/groups/_classifiers.py

@@ -26,7 +26,7 @@
 )
 from simcore_postgres_database.models.classifiers import group_classifiers
 
-from ..db.plugin import get_database_engine
+from ..db.plugin import get_aiopg_engine
 from ..scicrunch.db import ResearchResourceRepository
 from ..scicrunch.service_client import SciCrunch
 
@@ -75,7 +75,7 @@ class Classifiers(BaseModel):
 
 class GroupClassifierRepository:
     def __init__(self, app: web.Application):
-        self.engine = get_database_engine(app)
+        self.engine = get_aiopg_engine(app)
 
     async def _get_bundle(self, gid: int) -> RowProxy | None:
         async with self.engine.acquire() as conn:

services/web/server/src/simcore_service_webserver/groups/api.py

@@ -6,7 +6,7 @@
 from models_library.groups import Group
 from models_library.users import GroupID, UserID
 
-from ..db.plugin import get_database_engine
+from ..db.plugin import get_aiopg_engine
 from ..users.api import get_user
 from . import _db
 from ._utils import AccessRightsDict
@@ -21,15 +21,15 @@ async def list_user_groups_with_read_access(
     """
     # NOTE: Careful! It seems we are filtering out groups, such as Product Groups,
     # because they do not have read access. I believe this was done because the frontend did not want to display them.
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         return await _db.get_all_user_groups_with_read_access(conn, user_id=user_id)
 
 
 async def list_all_user_groups(app: web.Application, user_id: UserID) -> list[Group]:
     """
     Return all user groups
     """
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         groups_db = await _db.get_all_user_groups(conn, user_id=user_id)
 
     return [Group.construct(**group.dict()) for group in groups_db]
@@ -44,7 +44,7 @@ async def get_user_group(
     raises GroupNotFoundError
     raises UserInsufficientRightsError
     """
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         return await _db.get_user_group(conn, user_id=user_id, gid=gid)
 
 
@@ -55,7 +55,7 @@ async def get_product_group_for_user(
     Returns product's group if user belongs to it, otherwise it
     raises GroupNotFoundError
     """
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         return await _db.get_product_group_for_user(
             conn, user_id=user_id, product_gid=product_gid
         )
@@ -64,7 +64,7 @@ async def get_product_group_for_user(
 async def create_user_group(
     app: web.Application, user_id: UserID, new_group: dict
 ) -> dict[str, Any]:
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         return await _db.create_user_group(conn, user_id=user_id, new_group=new_group)
 
 
@@ -74,7 +74,7 @@ async def update_user_group(
     gid: GroupID,
     new_group_values: dict[str, str],
 ) -> dict[str, str]:
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         return await _db.update_user_group(
             conn, user_id=user_id, gid=gid, new_group_values=new_group_values
         )
@@ -83,28 +83,28 @@ async def update_user_group(
 async def delete_user_group(
     app: web.Application, user_id: UserID, gid: GroupID
 ) -> None:
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         return await _db.delete_user_group(conn, user_id=user_id, gid=gid)
 
 
 async def list_users_in_group(
     app: web.Application, user_id: UserID, gid: GroupID
 ) -> list[dict[str, str]]:
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         return await _db.list_users_in_group(conn, user_id=user_id, gid=gid)
 
 
 async def auto_add_user_to_groups(app: web.Application, user_id: UserID) -> None:
     user: dict = await get_user(app, user_id)
 
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         return await _db.auto_add_user_to_groups(conn, user=user)
 
 
 async def auto_add_user_to_product_group(
     app: web.Application, user_id: UserID, product_name: str
 ) -> GroupID:
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         return await _db.auto_add_user_to_product_group(
             conn, user_id=user_id, product_name=product_name
         )
@@ -113,7 +113,7 @@ async def auto_add_user_to_product_group(
 async def is_user_by_email_in_group(
     app: web.Application, user_email: LowerCaseEmailStr, group_id: GroupID
 ) -> bool:
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         return await _db.is_user_by_email_in_group(
             conn,
             email=user_email,
@@ -141,7 +141,7 @@ async def add_user_in_group(
         msg = "Invalid method call, missing user id or user email"
         raise GroupsError(msg)
 
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         if new_user_email:
             user: RowProxy = await _db.get_user_from_email(conn, new_user_email)
             new_user_id = user["id"]
@@ -162,7 +162,7 @@ async def add_user_in_group(
 async def get_user_in_group(
     app: web.Application, user_id: UserID, gid: GroupID, the_user_id_in_group: int
 ) -> dict[str, str]:
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         return await _db.get_user_in_group(
             conn, user_id=user_id, gid=gid, the_user_id_in_group=the_user_id_in_group
         )
@@ -175,7 +175,7 @@ async def update_user_in_group(
     the_user_id_in_group: int,
     new_values_for_user_in_group: dict,
 ) -> dict[str, str]:
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         return await _db.update_user_in_group(
             conn,
             user_id=user_id,
@@ -188,14 +188,14 @@ async def update_user_in_group(
 async def delete_user_in_group(
     app: web.Application, user_id: UserID, gid: GroupID, the_user_id_in_group: int
 ) -> None:
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         return await _db.delete_user_in_group(
             conn, user_id=user_id, gid=gid, the_user_id_in_group=the_user_id_in_group
         )
 
 
 async def get_group_from_gid(app: web.Application, gid: GroupID) -> Group | None:
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         group_db = await _db.get_group_from_gid(conn, gid=gid)
 
     if group_db:

services/web/server/src/simcore_service_webserver/login/_auth_api.py

@@ -6,7 +6,7 @@
 from simcore_postgres_database.models.users import UserStatus
 from simcore_postgres_database.utils_users import UsersRepo
 
-from ..db.plugin import get_database_engine
+from ..db.plugin import get_aiopg_engine
 from ..groups.api import is_user_by_email_in_group
 from ..products.api import Product
 from ..security.api import check_password, encrypt_password
@@ -30,7 +30,7 @@ async def create_user(
     expires_at: datetime | None,
 ) -> dict[str, Any]:
 
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         user = await UsersRepo.new_user(
             conn,
             email=email,

services/web/server/src/simcore_service_webserver/login/handlers_change.py

@@ -8,7 +8,7 @@
 from servicelib.mimetype_constants import MIMETYPE_APPLICATION_JSON
 from servicelib.request_keys import RQT_USERID_KEY
 from simcore_postgres_database.utils_users import UsersRepo
-from simcore_service_webserver.db.plugin import get_database_engine
+from simcore_service_webserver.db.plugin import get_aiopg_engine
 
 from .._meta import API_VTAG
 from ..products.api import Product, get_current_product
@@ -149,7 +149,7 @@ async def submit_request_to_change_email(request: web.Request):
     if user["email"] == request_body.email:
         return flash_response("Email changed")
 
-    async with get_database_engine(request.app).acquire() as conn:
+    async with get_aiopg_engine(request.app).acquire() as conn:
         if await UsersRepo.is_email_used(conn, email=request_body.email):
             raise web.HTTPUnprocessableEntity(reason="This email cannot be used")
 

services/web/server/src/simcore_service_webserver/payments/_autorecharge_db.py

@@ -9,7 +9,7 @@
 from pydantic import BaseModel, PositiveInt
 from simcore_postgres_database.utils_payments_autorecharge import AutoRechargeStmts
 
-from ..db.plugin import get_database_engine
+from ..db.plugin import get_aiopg_engine
 from .errors import InvalidPaymentMethodError
 
 _logger = logging.getLogger(__name__)
@@ -34,7 +34,7 @@ async def get_wallet_autorecharge(
     *,
     wallet_id: WalletID,
 ) -> PaymentsAutorechargeDB | None:
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         stmt = AutoRechargeStmts.get_wallet_autorecharge(wallet_id)
         result = await conn.execute(stmt)
         row = await result.first()
@@ -53,7 +53,7 @@ async def replace_wallet_autorecharge(
         InvalidPaymentMethodError: if `new` includes some invalid 'primary_payment_method_id'
 
     """
-    async with get_database_engine(app).acquire() as conn:
+    async with get_aiopg_engine(app).acquire() as conn:
         stmt = AutoRechargeStmts.is_valid_payment_method(
             user_id=user_id,
             wallet_id=new.wallet_id,