Implement rudimentary asyncio support w/ asyncpg

Using the approach introduced at
https://gist.github.com/zzzeek/6287e28054d3baddc07fa21a7227904e

We can now create asyncio endpoints that are then handled
in "implicit IO" form within the majority of the Core internals.
Then coroutines are re-exposed at the point at which we call
into asyncpg methods.

Patch includes:

* asyncpg dialect

* asyncio package

* engine, result, ORM session classes

* new test fixtures, tests

* some work with pep-484 and a short plugin for the
  pyannotate package, which seems to have so-so results

Change-Id: Idbcc0eff72c4cad572914acdd6f40ddb1aef1a7d
Fixes: #3414

1 files changed, 89 insertions, 0 deletions

diff --git a/examples/asyncio/async_orm.py b/examples/asyncio/async_orm.py
new file mode 100644
index 000000000..b1054a239
--- /dev/null
+++ b/examples/asyncio/async_orm.py
@@ -0,0 +1,89 @@
+"""Illustrates use of the sqlalchemy.ext.asyncio.AsyncSession object
+for asynchronous ORM use.
+
+"""
+
+import asyncio
+
+from sqlalchemy import Column
+from sqlalchemy import ForeignKey
+from sqlalchemy import Integer
+from sqlalchemy import String
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.future import select
+from sqlalchemy.orm import relationship
+from sqlalchemy.orm import selectinload
+
+Base = declarative_base()
+
+
+class A(Base):
+    __tablename__ = "a"
+
+    id = Column(Integer, primary_key=True)
+    data = Column(String)
+    bs = relationship("B")
+
+
+class B(Base):
+    __tablename__ = "b"
+    id = Column(Integer, primary_key=True)
+    a_id = Column(ForeignKey("a.id"))
+    data = Column(String)
+
+
+async def async_main():
+    """Main program function."""
+
+    engine = create_async_engine(
+        "postgresql+asyncpg://scott:tiger@localhost/test", echo=True,
+    )
+
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.drop_all)
+        await conn.run_sync(Base.metadata.create_all)
+
+    async with AsyncSession(engine) as session:
+        async with session.begin():
+            session.add_all(
+                [
+                    A(bs=[B(), B()], data="a1"),
+                    A(bs=[B()], data="a2"),
+                    A(bs=[B(), B()], data="a3"),
+                ]
+            )
+
+        # for relationship loading, eager loading should be applied.
+        stmt = select(A).options(selectinload(A.bs))
+
+        # AsyncSession.execute() is used for 2.0 style ORM execution
+        # (same as the synchronous API).
+        result = await session.execute(stmt)
+
+        # result is a buffered Result object.
+        for a1 in result.scalars():
+            print(a1)
+            for b1 in a1.bs:
+                print(b1)
+
+        # for streaming ORM results, AsyncSession.stream() may be used.
+        result = await session.stream(stmt)
+
+        # result is a streaming AsyncResult object.
+        async for a1 in result.scalars():
+            print(a1)
+            for b1 in a1.bs:
+                print(b1)
+
+        result = await session.execute(select(A).order_by(A.id))
+
+        a1 = result.scalars().first()
+
+        a1.data = "new data"
+
+        await session.commit()
+
+
+asyncio.run(async_main())