3 files changed, 16 insertions, 21 deletions
diff --git a/lib/sqlalchemy/ext/asyncio/base.py b/lib/sqlalchemy/ext/asyncio/base.py
index fa8c5006e..d11b059fd 100644
--- a/lib/sqlalchemy/ext/asyncio/base.py
+++ b/lib/sqlalchemy/ext/asyncio/base.py
@@ -5,14 +5,14 @@ from . import exc as async_exc
 
 class StartableContext(abc.ABC):
     @abc.abstractmethod
-    async def start(self) -> "StartableContext":
+    async def start(self, is_ctxmanager=False) -> "StartableContext":
         pass
 
     def __await__(self):
         return self.start().__await__()
 
     async def __aenter__(self):
-        return await self.start()
+        return await self.start(is_ctxmanager=True)
 
     @abc.abstractmethod
     async def __aexit__(self, type_, value, traceback):
diff --git a/lib/sqlalchemy/ext/asyncio/engine.py b/lib/sqlalchemy/ext/asyncio/engine.py
index c637b3d90..17ddb614a 100644
--- a/lib/sqlalchemy/ext/asyncio/engine.py
+++ b/lib/sqlalchemy/ext/asyncio/engine.py
@@ -101,7 +101,7 @@ class AsyncConnection(ProxyComparable, StartableContext, AsyncConnectable):
         self.sync_engine = async_engine.sync_engine
         self.sync_connection = sync_connection
 
-    async def start(self):
+    async def start(self, is_ctxmanager=False):
         """Start this :class:`_asyncio.AsyncConnection` object's context
         outside of using a Python ``with:`` block.
 
@@ -518,19 +518,15 @@ class AsyncEngine(ProxyComparable, AsyncConnectable):
         def __init__(self, conn):
             self.conn = conn
 
-        async def start(self):
-            await self.conn.start()
+        async def start(self, is_ctxmanager=False):
+            await self.conn.start(is_ctxmanager=is_ctxmanager)
             self.transaction = self.conn.begin()
             await self.transaction.__aenter__()
 
             return self.conn
 
         async def __aexit__(self, type_, value, traceback):
-            if type_ is not None:
-                await self.transaction.rollback()
-            else:
-                if self.transaction.is_active:
-                    await self.transaction.commit()
+            await self.transaction.__aexit__(type_, value, traceback)
             await self.conn.close()
 
     def __init__(self, sync_engine: Engine):
@@ -678,7 +674,7 @@ class AsyncTransaction(ProxyComparable, StartableContext):
 
         await greenlet_spawn(self._sync_transaction().commit)
 
-    async def start(self):
+    async def start(self, is_ctxmanager=False):
         """Start this :class:`_asyncio.AsyncTransaction` object's context
         outside of using a Python ``with:`` block.
 
@@ -689,17 +685,14 @@ class AsyncTransaction(ProxyComparable, StartableContext):
             if self.nested
             else self.connection._sync_connection().begin
         )
+        if is_ctxmanager:
+            self.sync_transaction.__enter__()
         return self
 
     async def __aexit__(self, type_, value, traceback):
-        if type_ is None and self.is_active:
-            try:
-                await self.commit()
-            except:
-                with util.safe_reraise():
-                    await self.rollback()
-        else:
-            await self.rollback()
+        await greenlet_spawn(
+            self._sync_transaction().__exit__, type_, value, traceback
+        )
 
 
 def _get_sync_engine_or_connection(async_engine):
diff --git a/lib/sqlalchemy/ext/asyncio/session.py b/lib/sqlalchemy/ext/asyncio/session.py
index d8a5673eb..1b61d6ee3 100644
--- a/lib/sqlalchemy/ext/asyncio/session.py
+++ b/lib/sqlalchemy/ext/asyncio/session.py
@@ -399,15 +399,17 @@ class AsyncSessionTransaction(StartableContext):
 
         await greenlet_spawn(self._sync_transaction().commit)
 
-    async def start(self):
+    async def start(self, is_ctxmanager=False):
         self.sync_transaction = await greenlet_spawn(
             self.session.sync_session.begin_nested
             if self.nested
             else self.session.sync_session.begin
         )
+        if is_ctxmanager:
+            self.sync_transaction.__enter__()
         return self
 
     async def __aexit__(self, type_, value, traceback):
-        return await greenlet_spawn(
+        await greenlet_spawn(
             self._sync_transaction().__exit__, type_, value, traceback
         )