feat(db_api): add an ability to set ReadOnly/ReadWrite connection mode

google/cloud/spanner_dbapi/connection.py

@@ -48,9 +48,14 @@ class Connection:
 
     :type database: :class:`~google.cloud.spanner_v1.database.Database`
     :param database: The database to which the connection is linked.
+
+    :type read_only: bool
+    :param read_only:
+        Flag to designate if the connection must use only ReadOnly
+        transaction type.
     """
 
-    def __init__(self, instance, database):
+    def __init__(self, instance, database, read_only=False):
         self._instance = instance
         self._database = database
         self._ddl_statements = []
@@ -67,6 +72,7 @@ def __init__(self, instance, database):
         # this connection should be cleared on the
         # connection close
         self._own_pool = True
+        self._read_only = read_only
 
     @property
     def autocommit(self):
@@ -121,6 +127,31 @@ def instance(self):
         """
         return self._instance
 
+    @property
+    def read_only(self):
+        """Flag: the connection can be used only for database reads.
+
+        Returns:
+            bool:
+                True, if the connection intended to be used for
+                database reads only.
+        """
+        return self._read_only
+
+    @read_only.setter
+    def read_only(self, value):
+        """`read_only` flag setter.
+
+        Args:
+            value (bool): True for ReadOnly mode, False for ReadWrite.
+        """
+        if self.inside_transaction:
+            raise ValueError(
+                "Connection read/write mode can't be changed while a transaction is in progress. "
+                "Commit or rollback the current transaction and try again."
+            )
+        self._read_only = value
+
     def _session_checkout(self):
         """Get a Cloud Spanner session from the pool.
 
@@ -209,7 +240,7 @@ def transaction_checkout(self):
         if not self.autocommit:
             if not self.inside_transaction:
                 self._transaction = self._session_checkout().transaction()
-                self._transaction.begin()
+                self._transaction.begin(read_only=self._read_only)
 
             return self._transaction
 
@@ -248,7 +279,12 @@ def commit(self):
         self.run_prior_DDL_statements()
         if self.inside_transaction:
             try:
-                self._transaction.commit()
+                if not self.read_only:
+                    self._transaction.commit()
+
+                self._session._transaction = None
+                self._transaction = None
+
                 self._release_session()
                 self._statements = []
             except Aborted:
@@ -264,7 +300,12 @@ def rollback(self):
         if self._autocommit:
             warnings.warn(AUTOCOMMIT_MODE_WARNING, UserWarning, stacklevel=2)
         elif self._transaction:
-            self._transaction.rollback()
+            if not self.read_only:
+                self._transaction.rollback()
+
+            self._session._transaction = None
+            self._transaction = None
+
             self._release_session()
             self._statements = []
 

google/cloud/spanner_v1/transaction.py

@@ -80,9 +80,14 @@ def _make_txn_selector(self):
         self._check_state()
         return TransactionSelector(id=self._transaction_id)
 
-    def begin(self):
+    def begin(self, read_only=False):
         """Begin a transaction on the database.
 
+        :type read_only: bool
+        :param read_only:
+            (Optional) If True, ReadOnly transaction type will be
+            begun, ReadWrite otherwise.
+
         :rtype: bytes
         :returns: the ID for the newly-begun transaction.
         :raises ValueError:
@@ -100,7 +105,10 @@ def begin(self):
         database = self._session._database
         api = database.spanner_api
         metadata = _metadata_with_prefix(database.name)
-        txn_options = TransactionOptions(read_write=TransactionOptions.ReadWrite())
+        if read_only:
+            txn_options = TransactionOptions(read_only=TransactionOptions.ReadOnly())
+        else:
+            txn_options = TransactionOptions(read_write=TransactionOptions.ReadWrite())
         with trace_call("CloudSpanner.BeginTransaction", self._session):
             response = api.begin_transaction(
                 session=self._session.name, options=txn_options, metadata=metadata

tests/system/test_system_dbapi.py

@@ -26,6 +26,7 @@
 from google.cloud.spanner_v1.instance import Instance
 
 from google.cloud.spanner_dbapi.connection import Connection
+from google.cloud.spanner_dbapi.exceptions import ProgrammingError
 
 from test_utils.retry import RetryErrors
 
@@ -426,6 +427,52 @@ def test_DDL_commit(self):
         cur.execute("DROP TABLE Singers")
         conn.commit()
 
+    def test_read_only(self):
+        """
+        Check that connection set to `read_only=True` uses
+        ReadOnly transactions.
+        """
+        conn = Connection(Config.INSTANCE, self._db, read_only=True)
+        cur = conn.cursor()
+
+        with self.assertRaises(ProgrammingError):
+            cur.execute(
+                """
+    UPDATE contacts
+    SET first_name = 'updated-first-name'
+    WHERE first_name = 'first-name'
+    """
+            )
+
+        cur.execute("SELECT * FROM contacts")
+
+        conn.commit()
+
+    def test_read_only_data_visibility(self):
+        conn_rw = Connection(Config.INSTANCE, self._db)
+        conn_ro = Connection(Config.INSTANCE, self._db, read_only=True)
+
+        cur_rw = conn_rw.cursor()
+        cur_ro = conn_ro.cursor()
+
+        # start read-only transaction
+        cur_ro.execute("SELECT * FROM contacts")
+
+        cur_rw.execute(
+            """INSERT INTO contacts (contact_id, first_name) VALUES (123, 'Inserted_while_read')"""
+        )
+        conn_rw.commit()
+
+        # try to read data inserted in parallel transaction
+        cur_ro.execute("SELECT * FROM contacts")
+        self.assertEqual(cur_ro.fetchall(), [])
+
+        # start new read-only transaction
+        conn_ro.commit()
+        cur_ro.execute("SELECT * FROM contacts")
+
+        self.assertEqual(cur_ro.fetchall(), [(123, "Inserted_while_read", None, None)])
+
 
 def clear_table(transaction):
     """Clear the test table."""

tests/unit/spanner_dbapi/test_connection.py

@@ -39,14 +39,14 @@ def _get_client_info(self):
 
         return ClientInfo(user_agent=USER_AGENT)
 
-    def _make_connection(self):
+    def _make_connection(self, **kwargs):
         from google.cloud.spanner_dbapi import Connection
         from google.cloud.spanner_v1.instance import Instance
 
         # We don't need a real Client object to test the constructor
         instance = Instance(INSTANCE, client=None)
         database = instance.database(DATABASE)
-        return Connection(instance, database)
+        return Connection(instance, database, **kwargs)
 
     @mock.patch("google.cloud.spanner_dbapi.connection.Connection.commit")
     def test_autocommit_setter_transaction_not_started(self, mock_commit):
@@ -105,6 +105,46 @@ def test_property_instance(self):
         self.assertIsInstance(connection.instance, Instance)
         self.assertEqual(connection.instance, connection._instance)
 
+    def test_read_only_connection(self):
+        connection = self._make_connection(read_only=True)
+        self.assertTrue(connection.read_only)
+
+        connection._transaction = mock.Mock(committed=False, rolled_back=False)
+        with self.assertRaisesRegex(
+            ValueError,
+            "Connection read/write mode can't be changed while a transaction is in progress. "
+            "Commit or rollback the current transaction and try again.",
+        ):
+            connection.read_only = False
+
+        connection._transaction = None
+        connection.read_only = False
+        self.assertFalse(connection.read_only)
+
+    def test_read_only_rollback_commit(self):
+        """
+        Check that ReadOnly transactions are not committed
+        or rolled back, but sessions are still released.
+        """
+        connection = self._make_connection(read_only=True)
+
+        transaction = mock.Mock(committed=False, rolled_back=False)
+        transaction.commit = mock.Mock()
+        transaction.rollback = mock.Mock()
+        connection._release_session = mock.Mock()
+
+        connection._transaction = transaction
+
+        connection.commit()
+        transaction.commit.assert_not_called()
+        connection._release_session.assert_called_once()
+
+        connection._release_session.reset_mock()
+
+        connection.rollback()
+        transaction.rollback.assert_not_called()
+        connection._release_session.assert_called_once()
+
     @staticmethod
     def _make_pool():
         from google.cloud.spanner_v1.pool import AbstractSessionPool

tests/unit/test_transaction.py

@@ -193,6 +193,31 @@ def test_begin_ok(self):
             "CloudSpanner.BeginTransaction", attributes=TestTransaction.BASE_ATTRIBUTES
         )
 
+    def test_begin_read_only(self):
+        from google.cloud.spanner_v1 import Transaction as TransactionPB
+
+        transaction_pb = TransactionPB(id=self.TRANSACTION_ID)
+        database = _Database()
+        api = database.spanner_api = _FauxSpannerAPI(
+            _begin_transaction_response=transaction_pb
+        )
+        session = _Session(database)
+        transaction = self._make_one(session)
+
+        txn_id = transaction.begin(read_only=True)
+
+        self.assertEqual(txn_id, self.TRANSACTION_ID)
+        self.assertEqual(transaction._transaction_id, self.TRANSACTION_ID)
+
+        session_id, txn_options, metadata = api._begun
+        self.assertEqual(session_id, session.name)
+        self.assertTrue(type(txn_options).pb(txn_options).HasField("read_only"))
+        self.assertEqual(metadata, [("google-cloud-resource-prefix", database.name)])
+
+        self.assertSpanAttributes(
+            "CloudSpanner.BeginTransaction", attributes=TestTransaction.BASE_ATTRIBUTES
+        )
+
     def test_rollback_not_begun(self):
         session = _Session()
         transaction = self._make_one(session)