Create connection characteristics API; implement postgresql flags

Added support for PostgreSQL "readonly" and "deferrable" flags for all of
psycopg2, asyncpg and pg8000 dialects.   This takes advantage of a newly
generalized version of the "isolation level" API to support other kinds of
session attributes set via execution options that are reliably reset
when connections are returned to the connection pool.

Fixes: #5549
Change-Id: I0ad6d7a095e49d331618274c40ce75c76afdc7dd

1 files changed, 42 insertions, 0 deletions

diff --git a/lib/sqlalchemy/dialects/postgresql/pg8000.py b/lib/sqlalchemy/dialects/postgresql/pg8000.py
index e08332a57..fd70828ff 100644
--- a/lib/sqlalchemy/dialects/postgresql/pg8000.py
+++ b/lib/sqlalchemy/dialects/postgresql/pg8000.py
@@ -359,6 +359,48 @@ class PGDialect_pg8000(PGDialect):
                 % (level, self.name, ", ".join(self._isolation_lookup))
             )
 
+    def set_readonly(self, connection, value):
+        cursor = connection.cursor()
+        try:
+            cursor.execute(
+                "SET SESSION CHARACTERISTICS AS TRANSACTION %s"
+                % ("READ ONLY" if value else "READ WRITE")
+            )
+            cursor.execute("COMMIT")
+        finally:
+            cursor.close()
+
+    def get_readonly(self, connection):
+        cursor = connection.cursor()
+        try:
+            cursor.execute("show transaction_read_only")
+            val = cursor.fetchone()[0]
+        finally:
+            cursor.close()
+
+        return val == "yes"
+
+    def set_deferrable(self, connection, value):
+        cursor = connection.cursor()
+        try:
+            cursor.execute(
+                "SET SESSION CHARACTERISTICS AS TRANSACTION %s"
+                % ("DEFERRABLE" if value else "NOT DEFERRABLE")
+            )
+            cursor.execute("COMMIT")
+        finally:
+            cursor.close()
+
+    def get_deferrable(self, connection):
+        cursor = connection.cursor()
+        try:
+            cursor.execute("show transaction_deferrable")
+            val = cursor.fetchone()[0]
+        finally:
+            cursor.close()
+
+        return val == "yes"
+
     def set_client_encoding(self, connection, client_encoding):
         # adjust for ConnectionFairy possibly being present
         if hasattr(connection, "connection"):