fixes test_base_repo

Author: pcrespov

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

@@ -27,10 +27,8 @@ async def transaction_context(
 ):
     async with get_or_create_connection(engine, connection) as conn:
         if conn.in_transaction():
-            # FIXME: should not extend nested? async with conn.begin_nested():
-            # depends on the analysis of test_sa_transactions
-            # might need another function that produces a transaction ONLY
-            yield conn
+            async with conn.begin_nested():  # savepoint
+                yield conn
         else:
             try:
                 async with conn.begin():

packages/postgres-database/tests/test_base_repo.py

@@ -4,7 +4,7 @@
 # pylint: disable=too-many-arguments
 
 
-from typing import Any, TypedDict
+from typing import Any, NamedTuple
 
 import pytest
 import sqlalchemy as sa
@@ -13,6 +13,7 @@
     transaction_context,
 )
 from simcore_postgres_database.models.tags import tags
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine
 
 
@@ -21,55 +22,57 @@ async def test_sa_transactions(asyncpg_engine: AsyncEngine):
     # SEE https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html#synopsis-core
     #
 
+    # READ query
     total_count_query = sa.select(sa.func.count()).select_from(tags)
 
     # WRITE queries
-    query1 = tags.insert().values(name="query1", color="blue")
-    query11 = tags.insert().values(name="query11", color="blue")
-    query111 = tags.insert().values(name="query111", color="blue")
-    query1111 = tags.insert().values(name="query1111", color="blue")
-    query112 = tags.insert().values(name="query112", color="blue")
-    query12 = tags.insert().values(name="query12", color="blue")
-    query2 = tags.insert().values(name="query2", color="blue")
+    query1 = (
+        tags.insert().values(id=2, name="query1", color="blue").returning(tags.c.id)
+    )
+    query11 = (
+        tags.insert().values(id=3, name="query11", color="blue").returning(tags.c.id)
+    )
+    query12 = (
+        tags.insert().values(id=5, name="query12", color="blue").returning(tags.c.id)
+    )
+    query2 = (
+        tags.insert().values(id=6, name="query2", color="blue").returning(tags.c.id)
+    )
+    query2 = (
+        tags.insert().values(id=7, name="query2", color="blue").returning(tags.c.id)
+    )
+
+    async with asyncpg_engine.connect() as conn, conn.begin():  # starts transaction (savepoint)
+
+        result = await conn.execute(query1)
+        assert result.scalar() == 2
 
-    # to make it fail, just repeat query since `id` is unique
-    # TODO: query1111 = tags.insert().values(id=1, name="query1111", color="blue")
-    # TODO: await conn.commit()  # explicit commit !
-
-    # TODO: if this is true, then the order of execution is NOT preserved?
-    async with asyncpg_engine.connect() as conn:
+        total_count = (await conn.execute(total_count_query)).scalar()
+        assert total_count == 1
 
-        await conn.execute(query1)
+        rows = (await conn.execute(tags.select().where(tags.c.id == 2))).fetchall()
+        assert rows
+        assert rows[0].id == 2
 
-        async with conn.begin():  # savepoint
+        async with conn.begin_nested():  # savepoint
             await conn.execute(query11)
 
-            async with conn.begin_nested():  # savepoint
-                await conn.execute(query111)
-
+            with pytest.raises(IntegrityError):
                 async with conn.begin_nested():  # savepoint
-                    await conn.execute(query1111)
-
-                await conn.execute(query112)
-
-                total_count = (await conn.execute(total_count_query)).scalar()
-                assert total_count == 1  #  (query1111)
+                    await conn.execute(query11)
 
             await conn.execute(query12)
 
             total_count = (await conn.execute(total_count_query)).scalar()
-            assert total_count == 3  #  query111, (query1111), query112
+            assert total_count == 3  # since query11 (second time) reverted!
 
         await conn.execute(query2)
 
         total_count = (await conn.execute(total_count_query)).scalar()
-        assert total_count == 5  #  query11, (query111, (query1111), query112), query2
-
-    total_count = (await conn.execute(total_count_query)).scalar()
-    assert total_count == 7  # includes query1, query2
+        assert total_count == 4
 
 
-class _PageDict(TypedDict):
+class _PageTuple(NamedTuple):
     total_count: int
     rows: list[dict[str, Any]]
 
@@ -95,22 +98,22 @@ async def get_by_id(
         self,
         connection: AsyncConnection | None = None,
         *,
-        record_id: int,
+        row_id: int,
     ) -> dict[str, Any] | None:
         async with get_or_create_connection(self.engine, connection) as conn:
             result = await conn.execute(
-                sa.select(self.table).where(self.table.c.id == record_id)
+                sa.select(self.table).where(self.table.c.id == row_id)
             )
-            record = result.fetchone()
-            return dict(record) if record else None
+            row = result.mappings().fetchone()
+            return dict(row) if row else None
 
     async def get_page(
         self,
         connection: AsyncConnection | None = None,
         *,
         limit: int,
-        offset: int,
-    ) -> _PageDict:
+        offset: int = 0,
+    ) -> _PageTuple:
         async with get_or_create_connection(self.engine, connection) as conn:
             # Compute total count
             total_count_query = sa.select(sa.func.count()).select_from(self.table)
@@ -120,81 +123,91 @@ async def get_page(
             # Fetch paginated results
             query = sa.select(self.table).limit(limit).offset(offset)
             result = await conn.execute(query)
-            records = [dict(**row) for row in result.fetchall()]
+            rows = [dict(row) for row in result.mappings().fetchall()]
 
-            return _PageDict(total_count=total_count or 0, rows=records)
+            return _PageTuple(total_count=total_count or 0, rows=rows)
 
     async def update(
         self,
         connection: AsyncConnection | None = None,
         *,
-        record_id: int,
+        row_id: int,
         **values,
     ) -> bool:
         async with transaction_context(self.engine, connection) as conn:
             result = await conn.execute(
-                self.table.update().where(self.table.c.id == record_id).values(**values)
+                self.table.update().where(self.table.c.id == row_id).values(**values)
             )
             return result.rowcount > 0
 
     async def delete(
         self,
         connection: AsyncConnection | None = None,
         *,
-        record_id: int,
+        row_id: int,
     ) -> bool:
         async with transaction_context(self.engine, connection) as conn:
             result = await conn.execute(
-                self.table.delete().where(self.table.c.id == record_id)
+                self.table.delete().where(self.table.c.id == row_id)
             )
             return result.rowcount > 0
 
 
-async def test_transaction_context(asyncpg_engine: AsyncEngine):
+async def test_oneresourcerepodemo_prototype(asyncpg_engine: AsyncEngine):
 
     tags_repo = OneResourceRepoDemo(engine=asyncpg_engine, table=tags)
 
+    # create
+    tag_id = await tags_repo.create(name="cyan tag", color="cyan")
+    assert tag_id > 0
+
+    # get, list
+    tag = await tags_repo.get_by_id(row_id=tag_id)
+    assert tag
+
+    page = await tags_repo.get_page(limit=10)
+    assert page.total_count == 1
+    assert page.rows == [tag]
+
+    # update
+    ok = await tags_repo.update(row_id=tag_id, name="changed name")
+    assert ok
+
+    updated_tag = await tags_repo.get_by_id(row_id=tag_id)
+    assert updated_tag
+    assert updated_tag["name"] != tag["name"]
+
+    # delete
+    ok = await tags_repo.delete(row_id=tag_id)
+    assert ok
+
+    assert not await tags_repo.get_by_id(row_id=tag_id)
+
+
+async def test_transaction_context(asyncpg_engine: AsyncEngine):
     # (1) Using transaction_context and fails
     fake_error_msg = "some error"
 
     def _something_raises_here():
         raise RuntimeError(fake_error_msg)
 
-    async def _create_blue_like_tags(connection):
-        # NOTE: embedded transaction here!!!
-        async with transaction_context(asyncpg_engine, connection) as conn:
-            await tags_repo.create(conn, name="cyan tag", color="cyan")
-            _something_raises_here()
-            await tags_repo.create(conn, name="violet tag", color="violet")
-
-    async def _create_four_tags(connection):
-        await tags_repo.create(connection, name="red tag", color="red")
-        await _create_blue_like_tags(connection)
-        await tags_repo.create(connection, name="green tag", color="green")
-
-    with pytest.raises(RuntimeError, match=fake_error_msg):
-        async with transaction_context(asyncpg_engine) as conn:
-            await tags_repo.create(conn, name="red tag", color="red")
-            _something_raises_here()
-            await tags_repo.create(conn, name="green tag", color="green")
-
-    print(asyncpg_engine.pool.status())
-    assert conn.closed
+    tags_repo = OneResourceRepoDemo(engine=asyncpg_engine, table=tags)
 
-    page = await tags_repo.get_page(limit=50, offset=0)
-    assert page["total_count"] == 0, "Transaction did not happen"
+    # using external transaction_context: commits upon __aexit__
+    async with transaction_context(asyncpg_engine) as conn:
+        await tags_repo.create(conn, name="cyan tag", color="cyan")
+        await tags_repo.create(conn, name="red tag", color="red")
+        assert (await tags_repo.get_page(conn, limit=10, offset=0)).total_count == 2
 
-    # (2) using internal connections
-    await tags_repo.create(name="blue tag", color="blue")
+    # using internal: auto-commit
     await tags_repo.create(name="red tag", color="red")
-    page = await tags_repo.get_page(limit=50, offset=0)
-    assert page["total_count"] == 2
+    assert (await tags_repo.get_page(limit=10, offset=0)).total_count == 3
 
-    # (3) using external embedded
-    async with transaction_context(asyncpg_engine) as conn:
-        page = await tags_repo.get_page(conn, limit=50, offset=0)
-        assert page["total_count"] == 2
+    # auto-rollback
+    with pytest.raises(RuntimeError, match=fake_error_msg):  # noqa: PT012
+        async with transaction_context(asyncpg_engine) as conn:
+            await tags_repo.create(conn, name="violet tag", color="violet")
+            assert (await tags_repo.get_page(conn, limit=10, offset=0)).total_count == 4
+            _something_raises_here()
 
-        # select a Result, which will be delivered with buffered results
-        result = await conn.execute(sa.select(tags).where(tags.c.name == "blue tag"))
-        assert result.fetchall()
+    assert (await tags_repo.get_page(limit=10, offset=0)).total_count == 3