Merge "set autocommit for psycopg2 pre-ping"

3 files changed, 79 insertions, 30 deletions

diff --git a/doc/build/changelog/unreleased_14/6621.rst b/doc/build/changelog/unreleased_14/6621.rst
new file mode 100644
index 000000000..0a8720617
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/6621.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, postgresql
+    :tickets: 6621
+
+    Fixed issue where the pool "pre ping" feature would implicitly start a
+    transaction, which would then interfere with custom transactional flags
+    such as PostgreSQL's "read only" mode when used with the psycopg2 driver.
+
diff --git a/lib/sqlalchemy/dialects/postgresql/psycopg2.py b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
index 3ef01e9e9..19ca14265 100644
--- a/lib/sqlalchemy/dialects/postgresql/psycopg2.py
+++ b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
@@ -844,6 +844,25 @@ class PGDialect_psycopg2(PGDialect):
     def get_deferrable(self, connection):
         return connection.deferrable
 
+    def do_ping(self, dbapi_connection):
+        cursor = None
+        try:
+            dbapi_connection.autocommit = True
+            cursor = dbapi_connection.cursor()
+            try:
+                cursor.execute(self._dialect_specific_select_one)
+            finally:
+                cursor.close()
+                if not dbapi_connection.closed:
+                    dbapi_connection.autocommit = False
+        except self.dbapi.Error as err:
+            if self.is_disconnect(err, dbapi_connection, cursor):
+                return False
+            else:
+                raise
+        else:
+            return True
+
     def on_connect(self):
         extras = self._psycopg2_extras()
         extensions = self._psycopg2_extensions()
diff --git a/test/dialect/postgresql/test_dialect.py b/test/dialect/postgresql/test_dialect.py
index 69bb41e2c..5a53e0b7e 100644
--- a/test/dialect/postgresql/test_dialect.py
+++ b/test/dialect/postgresql/test_dialect.py
@@ -895,21 +895,30 @@ class MiscBackendTest(
                 txid2 = conn.exec_driver_sql("select txid_current()").scalar()
                 eq_(txid1, txid2)
 
-    def test_readonly_flag_connection(self):
-        with testing.db.connect() as conn:
-            # asyncpg requires serializable for readonly..
-            conn = conn.execution_options(
-                isolation_level="SERIALIZABLE", postgresql_readonly=True
-            )
+    @testing.combinations((True,), (False,), argnames="pre_ping")
+    def test_readonly_flag_connection(self, testing_engine, pre_ping):
+        if pre_ping:
+            engine = testing_engine(options={"pool_pre_ping": True})
+        else:
+            engine = testing_engine()
 
-            dbapi_conn = conn.connection.connection
+        for i in range(2):
+            with engine.connect() as conn:
+                # asyncpg requires serializable for readonly..
+                conn = conn.execution_options(
+                    isolation_level="SERIALIZABLE", postgresql_readonly=True
+                )
 
-            cursor = dbapi_conn.cursor()
-            cursor.execute("show transaction_read_only")
-            val = cursor.fetchone()[0]
-            cursor.close()
-            eq_(val, "on")
-            is_true(testing.db.dialect.get_readonly(dbapi_conn))
+                conn.execute(text("select 1")).scalar()
+
+                dbapi_conn = conn.connection.connection
+
+                cursor = dbapi_conn.cursor()
+                cursor.execute("show transaction_read_only")
+                val = cursor.fetchone()[0]
+                cursor.close()
+                eq_(val, "on")
+                is_true(testing.db.dialect.get_readonly(dbapi_conn))
 
         cursor = dbapi_conn.cursor()
         try:
@@ -920,22 +929,31 @@ class MiscBackendTest(
             dbapi_conn.rollback()
         eq_(val, "off")
 
-    def test_deferrable_flag_connection(self):
-        with testing.db.connect() as conn:
-            # asyncpg but not for deferrable?  which the PG docs actually
-            # state.  weird
-            conn = conn.execution_options(
-                isolation_level="SERIALIZABLE", postgresql_deferrable=True
-            )
+    @testing.combinations((True,), (False,), argnames="pre_ping")
+    def test_deferrable_flag_connection(self, testing_engine, pre_ping):
+        if pre_ping:
+            engine = testing_engine(options={"pool_pre_ping": True})
+        else:
+            engine = testing_engine()
 
-            dbapi_conn = conn.connection.connection
+        for i in range(2):
+            with engine.connect() as conn:
+                # asyncpg but not for deferrable?  which the PG docs actually
+                # state.  weird
+                conn = conn.execution_options(
+                    isolation_level="SERIALIZABLE", postgresql_deferrable=True
+                )
 
-            cursor = dbapi_conn.cursor()
-            cursor.execute("show transaction_deferrable")
-            val = cursor.fetchone()[0]
-            cursor.close()
-            eq_(val, "on")
-            is_true(testing.db.dialect.get_deferrable(dbapi_conn))
+                conn.execute(text("Select 1")).scalar()
+
+                dbapi_conn = conn.connection.connection
+
+                cursor = dbapi_conn.cursor()
+                cursor.execute("show transaction_deferrable")
+                val = cursor.fetchone()[0]
+                cursor.close()
+                eq_(val, "on")
+                is_true(testing.db.dialect.get_deferrable(dbapi_conn))
 
         cursor = dbapi_conn.cursor()
         try:
@@ -946,16 +964,20 @@ class MiscBackendTest(
             dbapi_conn.rollback()
         eq_(val, "off")
 
-    def test_readonly_flag_engine(self):
-        engine = engines.testing_engine(
+    @testing.combinations((True,), (False,), argnames="pre_ping")
+    def test_readonly_flag_engine(self, testing_engine, pre_ping):
+        engine = testing_engine(
             options={
                 "execution_options": dict(
                     isolation_level="SERIALIZABLE", postgresql_readonly=True
-                )
+                ),
+                "pool_pre_ping": pre_ping,
             }
         )
         for i in range(2):
             with engine.connect() as conn:
+                conn.execute(text("select 1")).scalar()
+
                 dbapi_conn = conn.connection.connection
 
                 cursor = dbapi_conn.cursor()