Implement async_scoped_session

Implemented :class:`_asyncio.async_scoped_session` to address some
asyncio-related incompatibilities between :class:`_orm.scoped_session` and
:class:`_asyncio.AsyncSession`, in which some methods (notably the
:meth:`_asyncio.async_scoped_session.remove` method) should be used with
the ``await`` keyword.

Fixes: #6583
Closes: #6603
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/6603
Pull-request-sha: 0e8ef87dc824dcd83dca01641441afc453c8e07a

Change-Id: I9bfe56f8670302ff0015d9dc56c1e3ac5b92b118

1 files changed, 44 insertions, 0 deletions

diff --git a/test/ext/asyncio/test_scoping_py3k.py b/test/ext/asyncio/test_scoping_py3k.py
new file mode 100644
index 000000000..223c7d903
--- /dev/null
+++ b/test/ext/asyncio/test_scoping_py3k.py
@@ -0,0 +1,44 @@
+from asyncio import current_task
+
+import sqlalchemy as sa
+from sqlalchemy import func
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import async_scoped_session
+from sqlalchemy.ext.asyncio import AsyncSession as _AsyncSession
+from sqlalchemy.testing import async_test
+from sqlalchemy.testing import eq_
+from sqlalchemy.testing import is_
+from .test_session_py3k import AsyncFixture
+
+
+class AsyncScopedSessionTest(AsyncFixture):
+    @async_test
+    async def test_basic(self, async_engine):
+        AsyncSession = async_scoped_session(
+            sa.orm.sessionmaker(async_engine, class_=_AsyncSession),
+            scopefunc=current_task,
+        )
+
+        some_async_session = AsyncSession()
+        some_other_async_session = AsyncSession()
+
+        is_(some_async_session, some_other_async_session)
+        is_(some_async_session.bind, async_engine)
+
+        User = self.classes.User
+
+        async with AsyncSession.begin():
+            user_name = "scoped_async_session_u1"
+            u1 = User(name=user_name)
+
+            AsyncSession.add(u1)
+
+            await AsyncSession.flush()
+
+            conn = await AsyncSession.connection()
+            stmt = select(func.count(User.id)).where(User.name == user_name)
+            eq_(await conn.scalar(stmt), 1)
+
+            await AsyncSession.delete(u1)
+            await AsyncSession.flush()
+            eq_(await conn.scalar(stmt), 0)