asyncpg version of aggregation

Author: pcrespov

packages/postgres-database/src/simcore_postgres_database/utils_groups_extra_properties.py

@@ -1,15 +1,18 @@
 import datetime
 import logging
+import warnings
 from dataclasses import dataclass, fields
 from typing import Any
 
 import sqlalchemy as sa
 from aiopg.sa.connection import SAConnection
 from aiopg.sa.result import RowProxy
+from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine
 
 from .models.groups import GroupType, groups, user_to_groups
 from .models.groups_extra_properties import groups_extra_properties
 from .utils_models import FromRowMixin
+from .utils_repos import pass_or_acquire_connection
 
 _logger = logging.getLogger(__name__)
 
@@ -103,31 +106,52 @@ def _merge_extra_properties_booleans(
 
 @dataclass(frozen=True, slots=True, kw_only=True)
 class GroupExtraPropertiesRepo:
+    @staticmethod
+    def _get_stmt(gid: int, product_name: str):
+        return sa.select(groups_extra_properties).where(
+            (groups_extra_properties.c.group_id == gid)
+            & (groups_extra_properties.c.product_name == product_name)
+        )
+
     @staticmethod
     async def get(
         connection: SAConnection, *, gid: int, product_name: str
     ) -> GroupExtraProperties:
-        get_stmt = sa.select(groups_extra_properties).where(
-            (groups_extra_properties.c.group_id == gid)
-            & (groups_extra_properties.c.product_name == product_name)
+        warnings.warn(
+            f"{__name__}.get_v2 uses aiopg which has been deprecated in this repo."
+            "Use get_v2 instead. "
+            "See https://github.com/ITISFoundation/osparc-simcore/issues/4529",
+            DeprecationWarning,
+            stacklevel=1,
         )
-        result = await connection.execute(get_stmt)
+
+        query = GroupExtraPropertiesRepo._get_stmt(gid, product_name)
+        result = await connection.execute(query)
         assert result  # nosec
         if row := await result.first():
             return GroupExtraProperties.from_row_proxy(row)
         msg = f"Properties for group {gid} not found"
         raise GroupExtraPropertiesNotFoundError(msg)
 
     @staticmethod
-    async def get_aggregated_properties_for_user(
-        connection: SAConnection,
+    async def get_v2(
+        engine: AsyncEngine,
+        connection: AsyncConnection | None = None,
         *,
-        user_id: int,
+        gid: int,
         product_name: str,
     ) -> GroupExtraProperties:
-        rows = await _list_table_entries_ordered_by_group_type(
-            connection, user_id, product_name
-        )
+        async with pass_or_acquire_connection(engine, connection) as conn:
+            query = GroupExtraPropertiesRepo._get_stmt(gid, product_name)
+            result = await conn.stream(query)
+            assert result  # nosec
+            if row := await result.first():
+                return GroupExtraProperties.from_orm(row)
+            msg = f"Properties for group {gid} not found"
+            raise GroupExtraPropertiesNotFoundError(msg)
+
+    @staticmethod
+    def _aggregate(rows, user_id, product_name):
         merged_standard_extra_properties = None
         for row in rows:
             group_extra_properties = GroupExtraProperties.from_row_proxy(row)
@@ -161,3 +185,42 @@ async def get_aggregated_properties_for_user(
             return merged_standard_extra_properties
         msg = f"Properties for user {user_id} in {product_name} not found"
         raise GroupExtraPropertiesNotFoundError(msg)
+
+    @staticmethod
+    async def get_aggregated_properties_for_user(
+        connection: SAConnection,
+        *,
+        user_id: int,
+        product_name: str,
+    ) -> GroupExtraProperties:
+        warnings.warn(
+            f"{__name__}.get_aggregated_properties_for_user uses aiopg which has been deprecated in this repo. "
+            "Use get_aggregated_properties_for_user_v2 instead. "
+            "See https://github.com/ITISFoundation/osparc-simcore/issues/4529",
+            DeprecationWarning,
+            stacklevel=1,
+        )
+
+        rows = await _list_table_entries_ordered_by_group_type(
+            connection, user_id, product_name
+        )
+        return GroupExtraPropertiesRepo._aggregate(rows, user_id, product_name)
+
+    @staticmethod
+    async def get_aggregated_properties_for_user_v2(
+        engine: AsyncEngine,
+        connection: AsyncConnection | None = None,
+        *,
+        user_id: int,
+        product_name: str,
+    ) -> GroupExtraProperties:
+        async with pass_or_acquire_connection(engine, connection) as conn:
+
+            list_stmt = _list_table_entries_ordered_by_group_type_query(
+                user_id=user_id, product_name=product_name
+            )
+            result = await conn.stream(
+                sa.select(list_stmt).order_by(list_stmt.c.type_order)
+            )
+            rows = [row async for row in result]
+            return GroupExtraPropertiesRepo._aggregate(rows, user_id, product_name)

packages/postgres-database/src/simcore_postgres_database/utils_models.py

@@ -17,7 +17,7 @@ def from_row_proxy(cls: type[ModelType], row: RowProxy) -> ModelType:
         return cls(**{k: v for k, v in row.items() if k in field_names})  # type: ignore[return-value]
 
     @classmethod
-    def from_row(cls: type[ModelType], row: Row) -> ModelType:
+    def from_orm(cls: type[ModelType], row: Row) -> ModelType:
         assert is_dataclass(cls)  # nosec
         field_names = [f.name for f in fields(cls)]
         return cls(**{k: v for k, v in row._asdict().items() if k in field_names})  # type: ignore[return-value]

services/web/server/src/simcore_service_webserver/users/_users_repository.py

@@ -77,14 +77,12 @@ async def list_user_permissions(
         name="override_services_specifications",
         allowed=False,
     )
+    engine = get_asyncpg_engine(app)
     with contextlib.suppress(GroupExtraPropertiesNotFoundError):
-        async with pass_or_acquire_connection(
-            get_asyncpg_engine(app), connection
-        ) as conn:
+        async with pass_or_acquire_connection(engine, connection) as conn:
             user_group_extra_properties = (
-                # TODO: adapt to asyncpg
-                await GroupExtraPropertiesRepo.get_aggregated_properties_for_user(
-                    conn, user_id=user_id, product_name=product_name
+                await GroupExtraPropertiesRepo.get_aggregated_properties_for_user_v2(
+                    engine, conn, user_id=user_id, product_name=product_name
                 )
             )
         override_services_specifications.allowed = (