integrate connection.terminate() for supporting dialects

Integrated support for asyncpg's ``terminate()`` method call for cases
where the connection pool is recycling a possibly timed-out connection,
where a connection is being garbage collected that wasn't gracefully
closed, as well as when the connection has been invalidated. This allows
asyncpg to abandon the connection without waiting for a response that may
incur long timeouts.

Fixes: #8419
Change-Id: Ia575af779d5733b483a72dff3690b8bbbad2bb05

1 files changed, 7 insertions, 0 deletions

diff --git a/lib/sqlalchemy/dialects/postgresql/asyncpg.py b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
index 6888959f0..a84bece4f 100644
--- a/lib/sqlalchemy/dialects/postgresql/asyncpg.py
+++ b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
@@ -793,6 +793,9 @@ class AsyncAdapt_asyncpg_connection(AdaptedConnection):
 
         self.await_(self._connection.close())
 
+    def terminate(self):
+        self._connection.terminate()
+
 
 class AsyncAdaptFallback_asyncpg_connection(AsyncAdapt_asyncpg_connection):
     __slots__ = ()
@@ -895,6 +898,7 @@ class PGDialect_asyncpg(PGDialect):
     supports_server_side_cursors = True
 
     render_bind_cast = True
+    has_terminate = True
 
     default_paramstyle = "format"
     supports_sane_multi_rowcount = False
@@ -981,6 +985,9 @@ class PGDialect_asyncpg(PGDialect):
     def get_deferrable(self, connection):
         return connection.deferrable
 
+    def do_terminate(self, dbapi_connection) -> None:
+        dbapi_connection.terminate()
+
     def create_connect_args(self, url):
         opts = url.translate_connect_args(username="user")