WIP: Mai/db base repo

packages/models-library/Makefile

@@ -110,14 +110,15 @@ erd-ServiceInput.svg: _erdantic
 DOWNLOADED_TEST_DATA_DIR = "$(CURDIR)/tests/data/.downloaded-ignore"
 
 .PHONY: _httpx
-_httpx: _check_venv_active
+_ensure_httpx: _check_venv_active
 	# ensures requirements installed
 	@python3 -c "import httpx" 2>/dev/null || uv pip install httpx
 
-PHONY: pull_test_data
-pull_test_data: $(DOT_ENV_FILE) _httpx ## downloads tests data from registry (this can take some time!)
-	# downloading all metadata files
+PHONY: tests-data
+tests-data: $(DOT_ENV_FILE) _ensure_httpx ## downloads tests data from registry defined in .env (time-intensive!)
+	# Downloading all metadata files ...
 	@set -o allexport; \
 	source $<; \
 	set +o allexport; \
 	python3 "$(PACKAGES_DIR)/pytest-simcore/src/pytest_simcore/helpers/docker_registry.py" $(DOWNLOADED_TEST_DATA_DIR)
+	@echo "Run now 'pytest -vv -m diagnostics tests'"

packages/notifications-library/src/notifications_library/templates/on_account_form.email.content.html

@@ -4,7 +4,7 @@
 <p>Dear Support team
 
 <p>
-  We have received the following request form for an account in {{ product.display_name }} from {{ host }}
+  We have received the following request form for an account in {{ product.display_name }} from <b>{{ host }}</b>
 </p>
 
 <pre>

packages/notifications-library/src/notifications_library/templates/on_account_form.email.content.txt

@@ -1,6 +1,6 @@
 Dear Support team,
 
-We have received the following request form for an account in {{ product.display_name }} from {{ host }}:
+We have received the following request form for an account in {{ product.display_name }} from **{{ host }}**:
 
 {{ dumps(request_form) }}
 

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

@@ -0,0 +1,43 @@
+import logging
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine
+
+_logger = logging.getLogger(__name__)
+
+
+@asynccontextmanager
+async def get_or_create_connection(
+    engine: AsyncEngine, connection: AsyncConnection | None = None
+) -> AsyncIterator[AsyncConnection]:
+    # NOTE: When connection is passed, the engine is actually not needed
+    # NOTE: Creator is responsible of closing connection
+    is_connection_created = connection is None
+    if is_connection_created:
+        connection = await engine.connect()
+    try:
+        assert connection  # nosec
+        yield connection
+    finally:
+        assert connection  # nosec
+        assert not connection.closed  # nosec
+        if is_connection_created and connection:
+            await connection.close()
+
+
+@asynccontextmanager
+async def transaction_context(
+    engine: AsyncEngine, connection: AsyncConnection | None = None
+):
+    async with get_or_create_connection(engine, connection) as conn:
+        if conn.in_transaction():
+            async with conn.begin_nested():  # inner transaction (savepoint)
+                yield conn
+        else:
+            try:
+                async with conn.begin():  # outer transaction (savepoint)
+                    yield conn
+            finally:
+                assert not conn.closed  # nosec
+                assert not conn.in_transaction()  # nosec

packages/postgres-database/src/simcore_postgres_database/models/products.py

@@ -41,8 +41,6 @@ class Vendor(TypedDict, total=False):
     invitation_url: str  # How to request a trial invitation? (if applies)
     invitation_form: bool  # If True, it takes precendence over invitation_url and asks the FE to show the form (if defined)
 
-    has_landing_page: bool  # Is Landing page enabled
-
     release_notes_url_template: str  # a template url where `{vtag}` will be replaced, eg: "http://example.com/{vtag}.md"
 
 

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

@@ -1,12 +1,11 @@
 """ Repository pattern, errors and data structures for models.tags
 """
 
-import itertools
-from dataclasses import dataclass
 from typing import TypedDict
 
-from aiopg.sa.connection import SAConnection
+from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine
 
+from .base_repo import get_or_create_connection, transaction_context
 from .utils_tags_sql import (
     count_users_with_access_rights_stmt,
     create_tag_stmt,
@@ -49,15 +48,16 @@ class TagDict(TypedDict, total=True):
     delete: bool
 
 
-@dataclass(frozen=True)
 class TagsRepo:
-    user_id: int  # Determines access-rights
+    def __init__(self, engine: AsyncEngine):
+        self.engine = engine
 
     async def access_count(
         self,
-        conn: SAConnection,
-        tag_id: int,
+        connection: AsyncConnection | None = None,
         *,
+        user_id: int,
+        tag_id: int,
         read: bool | None = None,
         write: bool | None = None,
         delete: bool | None = None,
@@ -66,20 +66,22 @@ async def access_count(
         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
         """
-        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
+        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
 
     #
     # CRUD operations
     #
 
     async def create(
         self,
-        conn: SAConnection,
+        connection: AsyncConnection | None = None,
         *,
+        user_id: int,
         name: str,
         color: str,
         description: str | None = None,  # =nullable
@@ -94,69 +96,127 @@ async def create(
         if description:
             values["description"] = description
 
-        async with conn.begin():
+        async with transaction_context(self.engine, connection) as conn:
             # insert new tag
             insert_stmt = create_tag_stmt(**values)
             result = await conn.execute(insert_stmt)
-            tag = await result.first()
+            tag = result.first()
             assert tag  # nosec
 
             # take tag ownership
             access_stmt = set_tag_access_rights_stmt(
                 tag_id=tag.id,
-                user_id=self.user_id,
+                user_id=user_id,
                 read=read,
                 write=write,
                 delete=delete,
             )
             result = await conn.execute(access_stmt)
-            access = await result.first()
-            assert access
-
-            return TagDict(itertools.chain(tag.items(), access.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
+            access = result.first()
+            assert access  # nosec
+
+            return TagDict(
+                id=tag.id,
+                name=tag.name,
+                description=tag.description,
+                color=tag.color,
+                read=access.read,
+                write=access.write,
+                delete=access.delete,
+            )
 
-    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 list_all(
+        self,
+        connection: AsyncConnection | None = None,
+        *,
+        user_id: int,
+    ) -> list[TagDict]:
+        async with get_or_create_connection(self.engine, connection) as conn:
+            stmt_list = list_tags_stmt(user_id=user_id)
+            result = await conn.stream(stmt_list)
+            return [
+                TagDict(
+                    id=row.id,
+                    name=row.name,
+                    description=row.description,
+                    color=row.color,
+                    read=row.read,
+                    write=row.write,
+                    delete=row.delete,
+                )
+                async for row in result
+            ]
+
+    async def get(
+        self,
+        connection: AsyncConnection | None = None,
+        *,
+        user_id: int,
+        tag_id: int,
+    ) -> TagDict:
+        stmt_get = get_tag_stmt(user_id=user_id, tag_id=tag_id)
+        async with get_or_create_connection(self.engine, connection) as conn:
+            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(
+                id=row.id,
+                name=row.name,
+                description=row.description,
+                color=row.color,
+                read=row.read,
+                write=row.write,
+                delete=row.delete,
+            )
 
     async def update(
         self,
-        conn: SAConnection,
+        connection: AsyncConnection | None = None,
+        *,
+        user_id: int,
         tag_id: int,
         **fields,
     ) -> TagDict:
-        updates = {
-            name: value
-            for name, value in fields.items()
-            if name in {"name", "color", "description"}
-        }
-
-        if not updates:
-            # no updates == get
-            return await self.get(conn, tag_id=tag_id)
-
-        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
-
-    async def delete(self, conn: SAConnection, tag_id: int) -> None:
-        stmt_delete = delete_tag_stmt(user_id=self.user_id, tag_id=tag_id)
+        async with transaction_context(self.engine, connection) as conn:
+            updates = {
+                name: value
+                for name, value in fields.items()
+                if name in {"name", "color", "description"}
+            }
+
+            if not updates:
+                # no updates == get
+                return await self.get(conn, user_id=user_id, tag_id=tag_id)
+
+            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)
+
+            return TagDict(
+                id=row.id,
+                name=row.name,
+                description=row.description,
+                color=row.color,
+                read=row.read,
+                write=row.write,
+                delete=row.delete,
+            )
 
-        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,
+        connection: AsyncConnection | None = None,
+        *,
+        user_id: int,
+        tag_id: int,
+    ) -> None:
+        stmt_delete = delete_tag_stmt(user_id=user_id, tag_id=tag_id)
+        async with transaction_context(self.engine, connection) as conn:
+            deleted = await conn.scalar(stmt_delete)
+            if not deleted:
+                msg = f"Could not delete {tag_id=}. Not found or insuficient access."
+                raise TagOperationNotAllowedError(msg)