adapting base tests

Author: pcrespov

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

@@ -1,20 +1,23 @@
+import logging
 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
 ):
-    close_conn = False
-    if connection is None:
+    # creator is responsible of closing connection
+    is_connection_created = connection is None
+    if is_connection_created:
         connection = await engine.connect()
-        close_conn = True
     try:
         yield connection
     finally:
-        if close_conn:
+        if is_connection_created:
             await connection.close()
 
 
@@ -24,8 +27,12 @@ async def transaction_context(
 ):
     async with get_or_create_connection(engine, connection) as conn:
         if conn.in_transaction():
-            async with conn.begin_nested():
-                yield conn
+            # async with conn.begin_nested():
+            yield conn
         else:
-            async with conn.begin():
-                yield conn
+            try:
+                async with conn.begin():
+                    yield conn
+            finally:
+                assert not conn.closed  # nosec
+                assert not conn.in_transaction()  # nosec

packages/postgres-database/tests/test_base_repo.py

@@ -35,7 +35,6 @@ def __init__(self, engine: AsyncEngine, table: sa.Table):
     async def create(self, connection: AsyncConnection | None = None, **kwargs) -> int:
         async with transaction_context(self.engine, connection) as conn:
             result = await conn.execute(self.table.insert().values(**kwargs))
-            await conn.commit()
             assert result  # nosec
             return result.inserted_primary_key[0]
 
@@ -68,7 +67,7 @@ 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()]
+            records = [dict(**row) for row in result.fetchall()]
 
             return _PageDict(total_count=total_count or 0, rows=records)
 
@@ -83,7 +82,6 @@ async def update(
             result = await conn.execute(
                 self.table.update().where(self.table.c.id == record_id).values(**values)
             )
-            await conn.commit()
             return result.rowcount > 0
 
     async def delete(
@@ -96,10 +94,16 @@ async def delete(
             result = await conn.execute(
                 self.table.delete().where(self.table.c.id == record_id)
             )
-            await conn.commit()
             return result.rowcount > 0
 
 
+# async def test_it(asyncpg_engine: AsyncEngine):
+
+#     async with asyncpg_engine.connect() as conn:
+#         async with conn.begin():
+#             conn.execute()
+
+
 async def test_transaction_context(asyncpg_engine: AsyncEngine):
     #
     # Similar to example in https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html#synopsis-core
@@ -113,27 +117,33 @@ async def test_transaction_context(asyncpg_engine: AsyncEngine):
     def _something_raises_here():
         raise RuntimeError(fake_error_msg)
 
-    async def _create_blue_like_tags(conn):
-        async with conn.begin():  # NOTE: embedded transaction here!
+    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(conn):
-        await tags_repo.create(conn, name="red tag", color="red")
-        await _create_blue_like_tags(conn)
-        await tags_repo.create(conn, name="green tag", color="green")
+    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 _create_four_tags(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
 
     page = await tags_repo.get_page(limit=50, offset=0)
     assert page["total_count"] == 0, "Transaction did not happen"
 
     # (2) using internal connections
     await tags_repo.create(name="blue tag", color="blue")
-    await tags_repo.create(conn, name="red tag", color="red")
+    await tags_repo.create(name="red tag", color="red")
     page = await tags_repo.get_page(limit=50, offset=0)
     assert page["total_count"] == 2
 
@@ -143,5 +153,5 @@ async def _create_four_tags(conn):
         assert page["total_count"] == 2
 
         # select a Result, which will be delivered with buffered results
-        result = await conn.execute(sa.select(tags).where(tags.c.name == "some name 1"))
+        result = await conn.execute(sa.select(tags).where(tags.c.name == "blue tag"))
         assert result.fetchall()