rm new db features

Author: pcrespov

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

@@ -2,12 +2,11 @@
 """
 
 import itertools
+from dataclasses import dataclass
 from typing import TypedDict
 
-from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine
+from aiopg.sa.connection import SAConnection
 
-from .base_repo import MinimalRepo, get_or_create_connection, transaction_context
-from .models.tags import tags as tags_table
 from .tags_sql import (
     count_users_with_access_rights_stmt,
     create_tag_stmt,
@@ -50,46 +49,43 @@ class TagDict(TypedDict, total=True):
     delete: bool
 
 
+@dataclass(frozen=True)
 class TagsRepo:
-    def __init__(self, engine: AsyncEngine):
-        self.engine = engine
-        self._impl = MinimalRepo(engine=engine, table=tags_table)
+    user_id: int  # Determines access-rights
 
     async def access_count(
         self,
-        user_id: int,
+        conn: SAConnection,
         tag_id: int,
+        *,
         read: bool | None = None,
         write: bool | None = None,
         delete: bool | None = None,
-        connection: AsyncConnection | None = None,
     ) -> int:
         """
         Returns 0 if tag does not match access
         Returns >0 if it does and represents the number of groups granting this access to the user
         """
-        async with get_or_create_connection(self.engine, connection) as conn:
-            count_stmt = count_users_with_access_rights_stmt(
-                user_id=user_id, tag_id=tag_id, read=read, write=write, delete=delete
-            )
-            permissions_count: int | None = await conn.scalar(count_stmt)
-            return permissions_count if permissions_count else 0
+        count_stmt = count_users_with_access_rights_stmt(
+            user_id=self.user_id, tag_id=tag_id, read=read, write=write, delete=delete
+        )
+        permissions_count: int | None = await conn.scalar(count_stmt)
+        return permissions_count if permissions_count else 0
 
     #
     # CRUD operations
     #
 
     async def create(
         self,
+        conn: SAConnection,
         *,
-        user_id: int,
         name: str,
         color: str,
         description: str | None = None,  # =nullable
         read: bool = True,
         write: bool = True,
         delete: bool = True,
-        connection: AsyncConnection | None = None,
     ) -> TagDict:
         values = {
             "name": name,
@@ -98,58 +94,44 @@ async def create(
         if description:
             values["description"] = description
 
-        async with transaction_context(self.engine, connection) as conn:
+        async with conn.begin():
             # insert new tag
             insert_stmt = create_tag_stmt(**values)
             result = await conn.execute(insert_stmt)
-            tag = result.first()
+            tag = await result.first()
             assert tag  # nosec
 
             # take tag ownership
             access_stmt = set_tag_access_rights_stmt(
                 tag_id=tag.id,
-                user_id=user_id,
+                user_id=self.user_id,
                 read=read,
                 write=write,
                 delete=delete,
             )
             result = await conn.execute(access_stmt)
-            access = result.first()
+            access = await result.first()
             assert access
 
             return TagDict(itertools.chain(tag.items(), access.items()))  # type: ignore
 
-    async def list_all(
-        self,
-        *,
-        user_id: int,
-        connection: AsyncConnection | None = None,
-    ) -> list[TagDict]:
-        async with get_or_create_connection(self.engine, connection) as conn:
-            stmt_list = list_tags_stmt(user_id=user_id)
-            return [TagDict(row.items()) async for row in conn.execute(stmt_list)]  # type: ignore
-
-    async def get(
-        self,
-        user_id: int,
-        tag_id: int,
-        connection: AsyncConnection | None = None,
-    ) -> TagDict:
-        async with get_or_create_connection(self.engine, connection) as conn:
-            stmt_get = get_tag_stmt(user_id=user_id, tag_id=tag_id)
-            result = await conn.execute(stmt_get)
-            row = result.first()
-            if not row:
-                msg = f"{tag_id=} not found: either no access or does not exists"
-                raise TagNotFoundError(msg)
-            return TagDict(row.items())  # type: ignore
+    async def list_all(self, conn: SAConnection) -> list[TagDict]:
+        stmt_list = list_tags_stmt(user_id=self.user_id)
+        return [TagDict(row.items()) async for row in conn.execute(stmt_list)]  # type: ignore
+
+    async def get(self, conn: SAConnection, tag_id: int) -> TagDict:
+        stmt_get = get_tag_stmt(user_id=self.user_id, tag_id=tag_id)
+        result = await conn.execute(stmt_get)
+        row = await result.first()
+        if not row:
+            msg = f"{tag_id=} not found: either no access or does not exists"
+            raise TagNotFoundError(msg)
+        return TagDict(row.items())  # type: ignore
 
     async def update(
         self,
-        *,
+        conn: SAConnection,
         tag_id: int,
-        user_id: int,
-        connection: AsyncConnection | None = None,
         **fields,
     ) -> TagDict:
         updates = {
@@ -160,28 +142,21 @@ async def update(
 
         if not updates:
             # no updates == get
-            return await self.get(user_id=user_id, tag_id=tag_id, connection=connection)
+            return await self.get(conn, tag_id=tag_id)
 
-        async with get_or_create_connection(self.engine, connection) as conn:
-            update_stmt = update_tag_stmt(user_id=user_id, tag_id=tag_id, **updates)
-            result = await conn.execute(update_stmt)
-            row = result.first()
-            if not row:
-                msg = f"{tag_id=} not updated: either no access or not found"
-                raise TagOperationNotAllowedError(msg)
+        update_stmt = update_tag_stmt(user_id=self.user_id, tag_id=tag_id, **updates)
+        result = await conn.execute(update_stmt)
+        row = await result.first()
+        if not row:
+            msg = f"{tag_id=} not updated: either no access or not found"
+            raise TagOperationNotAllowedError(msg)
 
-            return TagDict(row.items())  # type: ignore
+        return TagDict(row.items())  # type: ignore
 
-    async def delete(
-        self,
-        *,
-        user_id: int,
-        tag_id: int,
-        connection: AsyncConnection | None = None,
-    ) -> None:
-        async with get_or_create_connection(self.engine, connection) as conn:
-            stmt_delete = delete_tag_stmt(user_id=user_id, tag_id=tag_id)
-            deleted = await conn.scalar(stmt_delete)
-            if not deleted:
-                msg = f"Could not delete {tag_id=}. Not found or insuficient access."
-                raise TagOperationNotAllowedError(msg)
+    async def delete(self, conn: SAConnection, tag_id: int) -> None:
+        stmt_delete = delete_tag_stmt(user_id=self.user_id, tag_id=tag_id)
+
+        deleted = await conn.scalar(stmt_delete)
+        if not deleted:
+            msg = f"Could not delete {tag_id=}. Not found or insuficient access."
+            raise TagOperationNotAllowedError(msg)

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

@@ -172,7 +172,7 @@ def get_tags_for_project_stmt(*, project_index: int):
     )
 
 
-def add_tag_to_project_stmt(*, project_index: int, tag_id: int) -> None:
+def add_tag_to_project_stmt(*, project_index: int, tag_id: int):
     return (
         pg_insert(projects_tags)
         .values(
@@ -190,7 +190,7 @@ def get_tags_for_services_stmt(*, key: str, version: str):
     )
 
 
-def add_tag_to_services_stmt(*, key: str, version: str, tag_id: int) -> None:
+def add_tag_to_services_stmt(*, key: str, version: str, tag_id: int):
     return (
         pg_insert(services_tags)
         .values(