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

4 files changed, 258 insertions, 0 deletions

diff --git a/examples/asyncio/__init__.py b/examples/asyncio/__init__.py
new file mode 100644
index 000000000..c53120f54
--- /dev/null
+++ b/examples/asyncio/__init__.py
@@ -0,0 +1,6 @@
+"""
+Examples illustrating the asyncio engine feature of SQLAlchemy.
+
+.. autosource::
+
+"""
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())
diff --git a/examples/asyncio/basic.py b/examples/asyncio/basic.py
new file mode 100644
index 000000000..05cdd8a05
--- /dev/null
+++ b/examples/asyncio/basic.py
@@ -0,0 +1,71 @@
+"""Illustrates the asyncio engine / connection interface.
+
+In this example, we have an async engine created by
+:func:`_engine.create_async_engine`.   We then use it using await
+within a coroutine.
+
+"""
+
+
+import asyncio
+
+from sqlalchemy import Column
+from sqlalchemy import Integer
+from sqlalchemy import MetaData
+from sqlalchemy import String
+from sqlalchemy import Table
+from sqlalchemy.ext.asyncio import create_async_engine
+
+
+meta = MetaData()
+
+t1 = Table(
+    "t1", meta, Column("id", Integer, primary_key=True), Column("name", String)
+)
+
+
+async def async_main():
+    # engine is an instance of AsyncEngine
+    engine = create_async_engine(
+        "postgresql+asyncpg://scott:tiger@localhost/test", echo=True,
+    )
+
+    # conn is an instance of AsyncConnection
+    async with engine.begin() as conn:
+
+        # to support SQLAlchemy DDL methods as well as legacy functions, the
+        # AsyncConnection.run_sync() awaitable method will pass a "sync"
+        # version of the AsyncConnection object to any synchronous method,
+        # where synchronous IO calls will be transparently translated for
+        # await.
+        await conn.run_sync(meta.drop_all)
+        await conn.run_sync(meta.create_all)
+
+        # for normal statement execution, a traditional "await execute()"
+        # pattern is used.
+        await conn.execute(
+            t1.insert(), [{"name": "some name 1"}, {"name": "some name 2"}]
+        )
+
+    async with engine.connect() as conn:
+
+        # the default result object is the
+        # sqlalchemy.engine.Result object
+        result = await conn.execute(t1.select())
+
+        # the results are buffered so no await call is necessary
+        # for this case.
+        print(result.fetchall())
+
+        # for a streaming result that buffers only segments of the
+        # result at time, the AsyncConnection.stream() method is used.
+        # this returns a sqlalchemy.ext.asyncio.AsyncResult object.
+        async_result = await conn.stream(t1.select())
+
+        # this object supports async iteration and awaitable
+        # versions of methods like .all(), fetchmany(), etc.
+        async for row in async_result:
+            print(row)
+
+
+asyncio.run(async_main())
diff --git a/examples/asyncio/greenlet_orm.py b/examples/asyncio/greenlet_orm.py
new file mode 100644
index 000000000..e0b568c4b
--- /dev/null
+++ b/examples/asyncio/greenlet_orm.py
@@ -0,0 +1,92 @@
+"""Illustrates use of the sqlalchemy.ext.asyncio.AsyncSession object
+for asynchronous ORM use, including the optional run_sync() method.
+
+
+"""
+
+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
+
+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)
+
+
+def run_queries(session):
+    """A function written in "synchronous" style that will be invoked
+    within the asyncio event loop.
+
+    The session object passed is a traditional orm.Session object with
+    synchronous interface.
+
+    """
+
+    stmt = select(A)
+
+    result = session.execute(stmt)
+
+    for a1 in result.scalars():
+        print(a1)
+        # lazy loads
+        for b1 in a1.bs:
+            print(b1)
+
+    result = session.execute(select(A).order_by(A.id))
+
+    a1 = result.scalars().first()
+
+    a1.data = "new data"
+
+
+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"),
+                ]
+            )
+
+        # we have the option to run a function written in sync style
+        # within the AsyncSession.run_sync() method.  The function will
+        # be passed a synchronous-style Session object and the function
+        # can use traditional ORM patterns.
+        await session.run_sync(run_queries)
+
+        await session.commit()
+
+
+asyncio.run(async_main())