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

4 files changed, 158 insertions, 1 deletions

diff --git a/lib/sqlalchemy/dialects/postgresql/asyncpg.py b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
index 515ef6e28..eb87249b4 100644
--- a/lib/sqlalchemy/dialects/postgresql/asyncpg.py
+++ b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
@@ -465,6 +465,8 @@ class AsyncAdapt_asyncpg_connection:
         "dbapi",
         "_connection",
         "isolation_level",
+        "readonly",
+        "deferrable",
         "_transaction",
         "_started",
     )
@@ -475,6 +477,8 @@ class AsyncAdapt_asyncpg_connection:
         self.dbapi = dbapi
         self._connection = connection
         self.isolation_level = "read_committed"
+        self.readonly = False
+        self.deferrable = False
         self._transaction = None
         self._started = False
         self.await_(self._setup_type_codecs())
@@ -530,7 +534,9 @@ class AsyncAdapt_asyncpg_connection:
 
         try:
             self._transaction = self._connection.transaction(
-                isolation=self.isolation_level
+                isolation=self.isolation_level,
+                readonly=self.readonly,
+                deferrable=self.deferrable,
             )
             await self._transaction.start()
         except Exception as error:
@@ -763,6 +769,18 @@ class PGDialect_asyncpg(PGDialect):
 
         connection.set_isolation_level(level)
 
+    def set_readonly(self, connection, value):
+        connection.readonly = value
+
+    def get_readonly(self, connection):
+        return connection.readonly
+
+    def set_deferrable(self, connection, value):
+        connection.deferrable = value
+
+    def get_deferrable(self, connection):
+        return connection.deferrable
+
     def create_connect_args(self, url):
         opts = url.translate_connect_args(username="user")
         if "port" in opts:
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index 8eb116111..6550dd20d 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -152,12 +152,44 @@ Valid values for ``isolation_level`` include:
 
 .. seealso::
 
+    :ref:`postgresql_readonly_deferrable`
+
     :ref:`dbapi_autocommit`
 
     :ref:`psycopg2_isolation_level`
 
     :ref:`pg8000_isolation_level`
 
+.. _postgresql_readonly_deferrable:
+
+Setting READ ONLY / DEFERRABLE
+------------------------------
+
+Most PostgreSQL dialects support setting the "READ ONLY" and "DEFERRABLE"
+characteristics of the transaction, which is in addition to the isolation level
+setting. These two attributes can be established either in conjunction with or
+independently of the isolation level by passing the ``postgresql_readonly`` and
+``postgresql_deferrable`` flags with
+:meth:`_engine.Connection.execution_options`.  The example below illustrates
+passing the ``"SERIALIZABLE"`` isolation level at the same time as setting
+"READ ONLY" and "DEFERRABLE"::
+
+    with engine.connect() as conn:
+        conn = conn.execution_options(
+            isolation_level="SERIALIZABLE",
+            postgresql_readonly=True,
+            postgresql_deferrable=True
+        )
+        with conn.begin():
+            #  ... work with transaction
+
+Note that some DBAPIs such as asyncpg only support "readonly" with
+SERIALIZABLE isolation.
+
+.. versionadded:: 1.4 added support for the ``postgresql_readonly``
+   and ``postgresql_deferrable`` execution options.
+
+
 .. _postgresql_schema_reflection:
 
 Remote-Schema Table Introspection and PostgreSQL search_path
@@ -1037,6 +1069,7 @@ from ... import exc
 from ... import schema
 from ... import sql
 from ... import util
+from ...engine import characteristics
 from ...engine import default
 from ...engine import reflection
 from ...sql import coercions
@@ -2610,6 +2643,36 @@ class PGExecutionContext(default.DefaultExecutionContext):
         return AUTOCOMMIT_REGEXP.match(statement)
 
 
+class PGReadOnlyConnectionCharacteristic(
+    characteristics.ConnectionCharacteristic
+):
+    transactional = True
+
+    def reset_characteristic(self, dialect, dbapi_conn):
+        dialect.set_readonly(dbapi_conn, False)
+
+    def set_characteristic(self, dialect, dbapi_conn, value):
+        dialect.set_readonly(dbapi_conn, value)
+
+    def get_characteristic(self, dialect, dbapi_conn):
+        return dialect.get_readonly(dbapi_conn)
+
+
+class PGDeferrableConnectionCharacteristic(
+    characteristics.ConnectionCharacteristic
+):
+    transactional = True
+
+    def reset_characteristic(self, dialect, dbapi_conn):
+        dialect.set_deferrable(dbapi_conn, False)
+
+    def set_characteristic(self, dialect, dbapi_conn, value):
+        dialect.set_deferrable(dbapi_conn, value)
+
+    def get_characteristic(self, dialect, dbapi_conn):
+        return dialect.get_deferrable(dbapi_conn)
+
+
 class PGDialect(default.DefaultDialect):
     name = "postgresql"
     supports_alter = True
@@ -2645,6 +2708,16 @@ class PGDialect(default.DefaultDialect):
     implicit_returning = True
     full_returning = True
 
+    connection_characteristics = (
+        default.DefaultDialect.connection_characteristics
+    )
+    connection_characteristics = connection_characteristics.union(
+        {
+            "postgresql_readonly": PGReadOnlyConnectionCharacteristic(),
+            "postgresql_deferrable": PGDeferrableConnectionCharacteristic(),
+        }
+    )
+
     construct_arguments = [
         (
             schema.Index,
@@ -2762,6 +2835,18 @@ class PGDialect(default.DefaultDialect):
         cursor.close()
         return val.upper()
 
+    def set_readonly(self, connection, value):
+        raise NotImplementedError()
+
+    def get_readonly(self, connection):
+        raise NotImplementedError()
+
+    def set_deferrable(self, connection, value):
+        raise NotImplementedError()
+
+    def get_deferrable(self, connection):
+        raise NotImplementedError()
+
     def do_begin_twophase(self, connection, xid):
         self.do_begin(connection.connection)
 
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"):
diff --git a/lib/sqlalchemy/dialects/postgresql/psycopg2.py b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
index 2161b24fc..3cc62fc93 100644
--- a/lib/sqlalchemy/dialects/postgresql/psycopg2.py
+++ b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
@@ -803,6 +803,18 @@ class PGDialect_psycopg2(PGDialect):
 
         connection.set_isolation_level(level)
 
+    def set_readonly(self, connection, value):
+        connection.readonly = value
+
+    def get_readonly(self, connection):
+        return connection.readonly
+
+    def set_deferrable(self, connection, value):
+        connection.deferrable = value
+
+    def get_deferrable(self, connection):
+        return connection.deferrable
+
     def on_connect(self):
         extras = self._psycopg2_extras()
         extensions = self._psycopg2_extensions()