drafted repo for asyncio

Author: pcrespov

packages/postgres-database/tests/conftest.py

@@ -5,6 +5,7 @@
 
 import uuid
 from collections.abc import AsyncIterator, Awaitable, Callable, Iterator
+from pathlib import Path
 
 import aiopg.sa
 import aiopg.sa.exc
@@ -36,6 +37,7 @@
     user_to_groups,
     users,
 )
+from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
 
 pytest_plugins = [
     "pytest_simcore.pytest_global_environs",
@@ -47,7 +49,7 @@
 def postgres_service(docker_services, docker_ip, docker_compose_file) -> str:
     """Deploys postgres and service is responsive"""
     # container environment
-    with open(docker_compose_file) as fh:
+    with Path.open(docker_compose_file) as fh:
         config = yaml.safe_load(fh)
     environ = config["services"]["postgres"]["environment"]
 
@@ -80,6 +82,29 @@ def _make(is_async=True) -> Awaitable[Engine] | sa.engine.base.Engine:
     return _make
 
 
+@pytest.fixture
+def make_asyncio_engine(postgres_service: str) -> Callable[[bool], AsyncEngine]:
+    dsn = postgres_service.replace("postgresql://", "postgresql+asyncpg://")
+    minsize = 1
+    maxsize = 50
+
+    def _(echo: bool):
+        engine: AsyncEngine = create_async_engine(
+            dsn,
+            pool_size=minsize,
+            max_overflow=maxsize - minsize,
+            connect_args={
+                "server_settings": {"application_name": "postgres_database_tests"}
+            },
+            pool_pre_ping=True,  # https://docs.sqlalchemy.org/en/14/core/pooling.html#dealing-with-disconnects
+            future=True,  # this uses sqlalchemy 2.0 API, shall be removed when sqlalchemy 2.0 is released
+            echo=echo,
+        )
+        return engine
+
+    return _
+
+
 def is_postgres_responsive(dsn) -> bool:
     """Check if something responds to ``url``"""
     try:

packages/postgres-database/tests/test_base_repo.py

@@ -0,0 +1,48 @@
+# pylint: disable=redefined-outer-name
+# pylint: disable=unused-argument
+# pylint: disable=unused-variable
+# pylint: disable=too-many-arguments
+
+from collections.abc import AsyncIterator, Callable
+
+import sqlalchemy as sa
+from simcore_postgres_database.base_repo import MinimalRepo, transaction_context
+from sqlalchemy.ext.asyncio import AsyncEngine
+
+
+async def asyncio_engine(
+    make_asyncio_engine: Callable[[bool], AsyncEngine]
+) -> AsyncIterator[AsyncEngine]:
+    engine = make_asyncio_engine(echo=True)
+    try:
+        yield engine
+    except Exception:
+        # for AsyncEngine created in function scope, close and
+        # clean-up pooled connections
+        await engine.dispose()
+
+
+async def test_it(asyncio_engine: AsyncEngine):
+
+    meta = sa.MetaData()
+    t1 = sa.Table("t1", meta, sa.Column("name", sa.String(50), primary_key=True))
+
+    t1_repo = MinimalRepo(engine=asyncio_engine, table=t1)
+
+    async with transaction_context(asyncio_engine) as conn:
+        await conn.run_sync(meta.drop_all)
+        await conn.run_sync(meta.create_all)
+
+        await t1_repo.create(conn, name="some name 1")
+        await t1_repo.create(conn, name="some name 2")
+
+    async with transaction_context(asyncio_engine) as conn:
+
+        page = await t1_repo.get_page(limit=50, offset=0, connection=conn)
+
+        assert page["total_count"] == 2
+
+        # select a Result, which will be delivered with buffered
+        # results
+        result = await conn.execute(sa.select(t1).where(t1.c.name == "some name 1"))
+        print(result.fetchall())