2 files changed, 26 insertions, 6 deletions
diff --git a/lib/sqlalchemy/dialects/postgresql/asyncpg.py b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
index e542c77f4..424ed0d50 100644
--- a/lib/sqlalchemy/dialects/postgresql/asyncpg.py
+++ b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
@@ -615,6 +615,10 @@ class AsyncAdapt_asyncpg_connection:
         return prepared_stmt, attributes
 
     def _handle_exception(self, error):
+        if self._connection.is_closed():
+            self._transaction = None
+            self._started = False
+
         if not isinstance(error, AsyncAdapt_asyncpg_dbapi.Error):
             exception_mapping = self.dbapi._asyncpg_error_translate
 
@@ -669,15 +673,23 @@ class AsyncAdapt_asyncpg_connection:
 
     def rollback(self):
         if self._started:
-            self.await_(self._transaction.rollback())
-            self._transaction = None
-            self._started = False
+            try:
+                self.await_(self._transaction.rollback())
+            except Exception as error:
+                self._handle_exception(error)
+            finally:
+                self._transaction = None
+                self._started = False
 
     def commit(self):
         if self._started:
-            self.await_(self._transaction.commit())
-            self._transaction = None
-            self._started = False
+            try:
+                self.await_(self._transaction.commit())
+            except Exception as error:
+                self._handle_exception(error)
+            finally:
+                self._transaction = None
+                self._started = False
 
     def close(self):
         self.rollback()
diff --git a/lib/sqlalchemy/testing/fixtures.py b/lib/sqlalchemy/testing/fixtures.py
index dcdeee5c9..08cab051e 100644
--- a/lib/sqlalchemy/testing/fixtures.py
+++ b/lib/sqlalchemy/testing/fixtures.py
@@ -107,6 +107,14 @@ class TestBase(object):
         engines.testing_reaper._drop_testing_engines("fixture")
 
     @config.fixture()
+    def async_testing_engine(self, testing_engine):
+        def go(**kw):
+            kw["asyncio"] = True
+            return testing_engine(**kw)
+
+        return go
+
+    @config.fixture()
     def metadata(self, request):
         """Provide bound MetaData for a single test, dropping afterwards."""