From 4d7684e99bf608016443097c2999307a4ba6a9ac Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Mon, 2 Aug 2021 12:42:13 +0300 Subject: [PATCH 01/27] feat(db_api): add an ability to set ReadOnly/ReadWrite connection mode --- google/cloud/spanner_dbapi/connection.py | 35 +++++++++++++++++++-- google/cloud/spanner_v1/transaction.py | 12 +++++-- tests/system/test_system_dbapi.py | 28 +++++++++++++++++ tests/unit/spanner_dbapi/test_connection.py | 20 ++++++++++-- tests/unit/test_transaction.py | 25 +++++++++++++++ 5 files changed, 114 insertions(+), 6 deletions(-) diff --git a/google/cloud/spanner_dbapi/connection.py b/google/cloud/spanner_dbapi/connection.py index 926408c928..b46f84f05f 100644 --- a/google/cloud/spanner_dbapi/connection.py +++ b/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 diff --git a/google/cloud/spanner_v1/transaction.py b/google/cloud/spanner_v1/transaction.py index fce14eb60d..060ccb7253 100644 --- a/google/cloud/spanner_v1/transaction.py +++ b/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 diff --git a/tests/system/test_system_dbapi.py b/tests/system/test_system_dbapi.py index 6ca1029ae1..b1bbe68df5 100644 --- a/tests/system/test_system_dbapi.py +++ b/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,33 @@ 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.assertRaisesRegex( + ProgrammingError, + "400 DML statements can only be performed in a read-write transaction.", + ): + cur.execute( + """ + UPDATE contacts + SET first_name = 'updated-first-name' + WHERE first_name = 'first-name' + """ + ) + + cur.execute("SELECT * FROM contacts") + + with self.assertRaisesRegex( + exceptions.FailedPrecondition, "400 Cannot commit a read-only transaction." + ): + conn.commit() + def clear_table(transaction): """Clear the test table.""" diff --git a/tests/unit/spanner_dbapi/test_connection.py b/tests/unit/spanner_dbapi/test_connection.py index 48129dcc2f..972f3284b5 100644 --- a/tests/unit/spanner_dbapi/test_connection.py +++ b/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,22 @@ 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) + @staticmethod def _make_pool(): from google.cloud.spanner_v1.pool import AbstractSessionPool diff --git a/tests/unit/test_transaction.py b/tests/unit/test_transaction.py index d87821fa4a..74729c6e27 100644 --- a/tests/unit/test_transaction.py +++ b/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) From 990fe3e57f3548a60eb88da823da06330cc4c7af Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Mon, 2 Aug 2021 12:49:26 +0300 Subject: [PATCH 02/27] update the error message --- tests/system/test_system_dbapi.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/system/test_system_dbapi.py b/tests/system/test_system_dbapi.py index b1bbe68df5..d4da5a3ee8 100644 --- a/tests/system/test_system_dbapi.py +++ b/tests/system/test_system_dbapi.py @@ -437,7 +437,7 @@ def test_read_only(self): with self.assertRaisesRegex( ProgrammingError, - "400 DML statements can only be performed in a read-write transaction.", + "400 DML statements can only be performed in a read-write or partitioned-dml transaction. Current transaction type is ReadOnly.", ): cur.execute( """ From 2381c6dc3d1f3d56f353a7c44f6e82cac2c441ef Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Mon, 2 Aug 2021 12:51:46 +0300 Subject: [PATCH 03/27] update another error message --- tests/system/test_system_dbapi.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/system/test_system_dbapi.py b/tests/system/test_system_dbapi.py index d4da5a3ee8..192d3e2e3f 100644 --- a/tests/system/test_system_dbapi.py +++ b/tests/system/test_system_dbapi.py @@ -450,7 +450,8 @@ def test_read_only(self): cur.execute("SELECT * FROM contacts") with self.assertRaisesRegex( - exceptions.FailedPrecondition, "400 Cannot commit a read-only transaction." + exceptions.FailedPrecondition, + "400 Cannot commit or rollback a read-only or partitioned-dml transaction.", ): conn.commit() From 9cec921e94efdb861d184b8bae02945f07a123b6 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Mon, 2 Aug 2021 15:05:56 +0300 Subject: [PATCH 04/27] don't check error messages with regexes --- tests/system/test_system_dbapi.py | 10 ++-------- 1 file changed, 2 insertions(+), 8 deletions(-) diff --git a/tests/system/test_system_dbapi.py b/tests/system/test_system_dbapi.py index 192d3e2e3f..eb8a7b4e8a 100644 --- a/tests/system/test_system_dbapi.py +++ b/tests/system/test_system_dbapi.py @@ -435,10 +435,7 @@ def test_read_only(self): conn = Connection(Config.INSTANCE, self._db, read_only=True) cur = conn.cursor() - with self.assertRaisesRegex( - ProgrammingError, - "400 DML statements can only be performed in a read-write or partitioned-dml transaction. Current transaction type is ReadOnly.", - ): + with self.assertRaises(ProgrammingError): cur.execute( """ UPDATE contacts @@ -449,10 +446,7 @@ def test_read_only(self): cur.execute("SELECT * FROM contacts") - with self.assertRaisesRegex( - exceptions.FailedPrecondition, - "400 Cannot commit or rollback a read-only or partitioned-dml transaction.", - ): + with self.assertRaises(exceptions.FailedPrecondition): conn.commit() From d6783b8113bedfeb180e151f2c9ec428482375c5 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Thu, 5 Aug 2021 11:02:11 +0300 Subject: [PATCH 05/27] don't commit/rollback ReadOnly transactions --- google/cloud/spanner_dbapi/connection.py | 6 ++++-- tests/system/test_system_dbapi.py | 3 +-- tests/unit/spanner_dbapi/test_connection.py | 24 +++++++++++++++++++++ 3 files changed, 29 insertions(+), 4 deletions(-) diff --git a/google/cloud/spanner_dbapi/connection.py b/google/cloud/spanner_dbapi/connection.py index b46f84f05f..595475254c 100644 --- a/google/cloud/spanner_dbapi/connection.py +++ b/google/cloud/spanner_dbapi/connection.py @@ -279,7 +279,8 @@ 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._release_session() self._statements = [] except Aborted: @@ -295,7 +296,8 @@ 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._release_session() self._statements = [] diff --git a/tests/system/test_system_dbapi.py b/tests/system/test_system_dbapi.py index eb8a7b4e8a..30c1acee86 100644 --- a/tests/system/test_system_dbapi.py +++ b/tests/system/test_system_dbapi.py @@ -446,8 +446,7 @@ def test_read_only(self): cur.execute("SELECT * FROM contacts") - with self.assertRaises(exceptions.FailedPrecondition): - conn.commit() + conn.commit() def clear_table(transaction): diff --git a/tests/unit/spanner_dbapi/test_connection.py b/tests/unit/spanner_dbapi/test_connection.py index 972f3284b5..f39b7f750e 100644 --- a/tests/unit/spanner_dbapi/test_connection.py +++ b/tests/unit/spanner_dbapi/test_connection.py @@ -121,6 +121,30 @@ def test_read_only_connection(self): 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 From 131a6fc93d84231865f3cdbf13b086f306dffe14 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Thu, 5 Aug 2021 11:26:44 +0300 Subject: [PATCH 06/27] clear the transaction --- google/cloud/spanner_dbapi/connection.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/google/cloud/spanner_dbapi/connection.py b/google/cloud/spanner_dbapi/connection.py index 595475254c..c34102c549 100644 --- a/google/cloud/spanner_dbapi/connection.py +++ b/google/cloud/spanner_dbapi/connection.py @@ -281,6 +281,10 @@ def commit(self): try: if not self.read_only: self._transaction.commit() + + self._session._transaction = None + self._transaction = None + self._release_session() self._statements = [] except Aborted: @@ -298,6 +302,10 @@ def rollback(self): elif self._transaction: if not self.read_only: self._transaction.rollback() + + self._session._transaction = None + self._transaction = None + self._release_session() self._statements = [] From e1a3db2dcaf8d222559d6fe9ac0bbeb1d9939182 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Thu, 5 Aug 2021 12:08:16 +0300 Subject: [PATCH 07/27] add read-only transactions data visibility test --- tests/system/test_system_dbapi.py | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/tests/system/test_system_dbapi.py b/tests/system/test_system_dbapi.py index 30c1acee86..c67c73b9df 100644 --- a/tests/system/test_system_dbapi.py +++ b/tests/system/test_system_dbapi.py @@ -448,6 +448,31 @@ def test_read_only(self): 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.""" From 4bd65e7c8ba62ea545f9ca5f9842d9596286cac0 Mon Sep 17 00:00:00 2001 From: Ilya Gurov Date: Thu, 5 Aug 2021 22:11:18 +0300 Subject: [PATCH 08/27] Apply suggestions from code review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Knut Olav Løite --- google/cloud/spanner_dbapi/connection.py | 17 +++++++++++++---- google/cloud/spanner_v1/transaction.py | 4 ++-- 2 files changed, 15 insertions(+), 6 deletions(-) diff --git a/google/cloud/spanner_dbapi/connection.py b/google/cloud/spanner_dbapi/connection.py index c34102c549..45a6918be5 100644 --- a/google/cloud/spanner_dbapi/connection.py +++ b/google/cloud/spanner_dbapi/connection.py @@ -51,8 +51,18 @@ class Connection: :type read_only: bool :param read_only: - Flag to designate if the connection must use only ReadOnly - transaction type. + Flag to indicate that the connection may only execute queries and no update or DDL statements. + If autocommit is true, the connection will use a single use read-only transaction with strong timestamp + bound for each new statement, and will immediately see any changes that have been committed by + any other transaction. + If autocommit is false, the connection will automatically start a new multi use read-only transaction + with strong timestamp bound when the first statement is executed. This read-only transaction will be + used for all subsequent statements until either commit() or rollback() is called on the connection. The + read-only transaction will read from a consistent snapshot of the database at the time that the + transaction started. This means that the transaction will not see any changes that have been + committed by other transactions since the start of the read-only transaction. Commit or rolling back + the read-only transaction is semantically the same, and only indicates that the read-only transaction + should end a that a new one should be started when the next statement is executed. """ def __init__(self, instance, database, read_only=False): @@ -133,8 +143,7 @@ def read_only(self): Returns: bool: - True, if the connection intended to be used for - database reads only. + True if the connection may only be used for database reads. """ return self._read_only diff --git a/google/cloud/spanner_v1/transaction.py b/google/cloud/spanner_v1/transaction.py index 060ccb7253..12ca80af6a 100644 --- a/google/cloud/spanner_v1/transaction.py +++ b/google/cloud/spanner_v1/transaction.py @@ -85,8 +85,8 @@ def begin(self, read_only=False): :type read_only: bool :param read_only: - (Optional) If True, ReadOnly transaction type will be - begun, ReadWrite otherwise. + (Optional) If True, a read-only transaction with strong timestamp bound will be started. + Otherwise a read/write transaction will be started. :rtype: bytes :returns: the ID for the newly-begun transaction. From 41866099c8c05b52a5ff943fdce53063c1650dd4 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Fri, 6 Aug 2021 11:01:04 +0300 Subject: [PATCH 09/27] add conditions for edge cases --- google/cloud/spanner_dbapi/connection.py | 6 ++++-- tests/unit/spanner_dbapi/test_connection.py | 1 + 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/google/cloud/spanner_dbapi/connection.py b/google/cloud/spanner_dbapi/connection.py index c34102c549..ec6f24bc6d 100644 --- a/google/cloud/spanner_dbapi/connection.py +++ b/google/cloud/spanner_dbapi/connection.py @@ -282,7 +282,8 @@ def commit(self): if not self.read_only: self._transaction.commit() - self._session._transaction = None + if self._session is not None: + self._session._transaction = None self._transaction = None self._release_session() @@ -303,7 +304,8 @@ def rollback(self): if not self.read_only: self._transaction.rollback() - self._session._transaction = None + if self._session is not None: + self._session._transaction = None self._transaction = None self._release_session() diff --git a/tests/unit/spanner_dbapi/test_connection.py b/tests/unit/spanner_dbapi/test_connection.py index f39b7f750e..c45c99f61d 100644 --- a/tests/unit/spanner_dbapi/test_connection.py +++ b/tests/unit/spanner_dbapi/test_connection.py @@ -140,6 +140,7 @@ def test_read_only_rollback_commit(self): connection._release_session.assert_called_once() connection._release_session.reset_mock() + connection._transaction = transaction connection.rollback() transaction.rollback.assert_not_called() From 8332d4977e658d7a47971932c1d19502bcb9d7fb Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Fri, 6 Aug 2021 11:34:07 +0300 Subject: [PATCH 10/27] don't calc checksum for read-only transactions --- google/cloud/spanner_dbapi/connection.py | 8 +++++--- google/cloud/spanner_dbapi/cursor.py | 6 +++--- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/google/cloud/spanner_dbapi/connection.py b/google/cloud/spanner_dbapi/connection.py index 28dc15205b..5e32c19f05 100644 --- a/google/cloud/spanner_dbapi/connection.py +++ b/google/cloud/spanner_dbapi/connection.py @@ -217,7 +217,7 @@ def _rerun_previous_statements(self): for statement in self._statements: res_iter, retried_checksum = self.run_statement(statement, retried=True) # executing all the completed statements - if statement != self._statements[-1]: + if statement != self._statements[-1] and not self.read_only: for res in res_iter: retried_checksum.consume_result(res) @@ -229,11 +229,13 @@ def _rerun_previous_statements(self): while len(retried_checksum) < len(statement.checksum): try: res = next(iter(res_iter)) - retried_checksum.consume_result(res) + if not self.read_only: + retried_checksum.consume_result(res) except StopIteration: break - _compare_checksums(statement.checksum, retried_checksum) + if not self.read_only: + _compare_checksums(statement.checksum, retried_checksum) def transaction_checkout(self): """Get a Cloud Spanner transaction. diff --git a/google/cloud/spanner_dbapi/cursor.py b/google/cloud/spanner_dbapi/cursor.py index c5de13b370..2f3c9fc704 100644 --- a/google/cloud/spanner_dbapi/cursor.py +++ b/google/cloud/spanner_dbapi/cursor.py @@ -272,7 +272,7 @@ def fetchone(self): try: res = next(self) - if not self.connection.autocommit: + if not self.connection.autocommit and not self.connection.read_only: self._checksum.consume_result(res) return res except StopIteration: @@ -290,7 +290,7 @@ def fetchall(self): res = [] try: for row in self: - if not self.connection.autocommit: + if not self.connection.autocommit and not self.connection.read_only: self._checksum.consume_result(row) res.append(row) except Aborted: @@ -319,7 +319,7 @@ def fetchmany(self, size=None): for i in range(size): try: res = next(self) - if not self.connection.autocommit: + if not self.connection.autocommit and not self.connection.read_only: self._checksum.consume_result(res) items.append(res) except StopIteration: From 0be92af79018a3d9560efd5c991ec6c7c420546f Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Tue, 10 Aug 2021 11:05:29 +0300 Subject: [PATCH 11/27] use Snapshot for reads --- google/cloud/spanner_dbapi/connection.py | 10 +--------- google/cloud/spanner_dbapi/cursor.py | 4 ++++ google/cloud/spanner_v1/transaction.py | 12 ++---------- tests/system/test_system_dbapi.py | 2 +- tests/unit/test_transaction.py | 25 ------------------------ 5 files changed, 8 insertions(+), 45 deletions(-) diff --git a/google/cloud/spanner_dbapi/connection.py b/google/cloud/spanner_dbapi/connection.py index 5e32c19f05..e195356681 100644 --- a/google/cloud/spanner_dbapi/connection.py +++ b/google/cloud/spanner_dbapi/connection.py @@ -251,7 +251,7 @@ def transaction_checkout(self): if not self.autocommit: if not self.inside_transaction: self._transaction = self._session_checkout().transaction() - self._transaction.begin(read_only=self._read_only) + self._transaction.begin() return self._transaction @@ -293,10 +293,6 @@ def commit(self): if not self.read_only: self._transaction.commit() - if self._session is not None: - self._session._transaction = None - self._transaction = None - self._release_session() self._statements = [] except Aborted: @@ -315,10 +311,6 @@ def rollback(self): if not self.read_only: self._transaction.rollback() - if self._session is not None: - self._session._transaction = None - self._transaction = None - self._release_session() self._statements = [] diff --git a/google/cloud/spanner_dbapi/cursor.py b/google/cloud/spanner_dbapi/cursor.py index 2f3c9fc704..0bb77db2d8 100644 --- a/google/cloud/spanner_dbapi/cursor.py +++ b/google/cloud/spanner_dbapi/cursor.py @@ -174,6 +174,10 @@ def execute(self, sql, args=None): # Classify whether this is a read-only SQL statement. try: + if self.connection.read_only: + self._handle_DQL(sql, args or None) + return + classification = parse_utils.classify_stmt(sql) if classification == parse_utils.STMT_DDL: ddl_statements = [] diff --git a/google/cloud/spanner_v1/transaction.py b/google/cloud/spanner_v1/transaction.py index 12ca80af6a..fce14eb60d 100644 --- a/google/cloud/spanner_v1/transaction.py +++ b/google/cloud/spanner_v1/transaction.py @@ -80,14 +80,9 @@ def _make_txn_selector(self): self._check_state() return TransactionSelector(id=self._transaction_id) - def begin(self, read_only=False): + def begin(self): """Begin a transaction on the database. - :type read_only: bool - :param read_only: - (Optional) If True, a read-only transaction with strong timestamp bound will be started. - Otherwise a read/write transaction will be started. - :rtype: bytes :returns: the ID for the newly-begun transaction. :raises ValueError: @@ -105,10 +100,7 @@ def begin(self, read_only=False): database = self._session._database api = database.spanner_api metadata = _metadata_with_prefix(database.name) - if read_only: - txn_options = TransactionOptions(read_only=TransactionOptions.ReadOnly()) - else: - txn_options = TransactionOptions(read_write=TransactionOptions.ReadWrite()) + 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 diff --git a/tests/system/test_system_dbapi.py b/tests/system/test_system_dbapi.py index c67c73b9df..c3b04725c5 100644 --- a/tests/system/test_system_dbapi.py +++ b/tests/system/test_system_dbapi.py @@ -465,7 +465,7 @@ def test_read_only_data_visibility(self): # try to read data inserted in parallel transaction cur_ro.execute("SELECT * FROM contacts") - self.assertEqual(cur_ro.fetchall(), []) + self.assertEqual(cur_ro.fetchall(), [(123, "Inserted_while_read", None, None)]) # start new read-only transaction conn_ro.commit() diff --git a/tests/unit/test_transaction.py b/tests/unit/test_transaction.py index 74729c6e27..d87821fa4a 100644 --- a/tests/unit/test_transaction.py +++ b/tests/unit/test_transaction.py @@ -193,31 +193,6 @@ 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) From 3a72694b8236be638bd391fbcdbe5c4e94d8d1b4 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Tue, 10 Aug 2021 11:07:05 +0300 Subject: [PATCH 12/27] update docstrings --- google/cloud/spanner_dbapi/connection.py | 10 +--------- 1 file changed, 1 insertion(+), 9 deletions(-) diff --git a/google/cloud/spanner_dbapi/connection.py b/google/cloud/spanner_dbapi/connection.py index e195356681..a912adb9a0 100644 --- a/google/cloud/spanner_dbapi/connection.py +++ b/google/cloud/spanner_dbapi/connection.py @@ -52,17 +52,9 @@ class Connection: :type read_only: bool :param read_only: Flag to indicate that the connection may only execute queries and no update or DDL statements. - If autocommit is true, the connection will use a single use read-only transaction with strong timestamp + If True, the connection will use a single use read-only transaction with strong timestamp bound for each new statement, and will immediately see any changes that have been committed by any other transaction. - If autocommit is false, the connection will automatically start a new multi use read-only transaction - with strong timestamp bound when the first statement is executed. This read-only transaction will be - used for all subsequent statements until either commit() or rollback() is called on the connection. The - read-only transaction will read from a consistent snapshot of the database at the time that the - transaction started. This means that the transaction will not see any changes that have been - committed by other transactions since the start of the read-only transaction. Commit or rolling back - the read-only transaction is semantically the same, and only indicates that the read-only transaction - should end a that a new one should be started when the next statement is executed. """ def __init__(self, instance, database, read_only=False): From de0e47eef47f371dc62a4f74c140518db0dc0e84 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Wed, 11 Aug 2021 11:45:18 +0300 Subject: [PATCH 13/27] use multi-use snapshots in !autocommit mode --- google/cloud/spanner_dbapi/connection.py | 19 ++++++ google/cloud/spanner_dbapi/cursor.py | 70 ++++++++++++--------- tests/unit/spanner_dbapi/test_connection.py | 21 +++++++ 3 files changed, 80 insertions(+), 30 deletions(-) diff --git a/google/cloud/spanner_dbapi/connection.py b/google/cloud/spanner_dbapi/connection.py index 4f72652599..7c8399c68a 100644 --- a/google/cloud/spanner_dbapi/connection.py +++ b/google/cloud/spanner_dbapi/connection.py @@ -21,6 +21,7 @@ from google.api_core.gapic_v1.client_info import ClientInfo from google.cloud import spanner_v1 as spanner from google.cloud.spanner_v1.session import _get_retry_delay +from google.cloud.spanner_v1.snapshot import Snapshot from google.cloud.spanner_dbapi._helpers import _execute_insert_heterogenous from google.cloud.spanner_dbapi._helpers import _execute_insert_homogenous @@ -66,6 +67,7 @@ def __init__(self, instance, database, read_only=False): self._transaction = None self._session = None + self._snapshot = None # SQL statements, which were executed # within the current transaction self._statements = [] @@ -266,6 +268,19 @@ def transaction_checkout(self): return self._transaction + def snapshot_checkout(self): + """Get a Cloud Spanner snapshot. + + Initiate a new multi-use snapshot, if there is no snapshot in + this connection yet. Return the existing one otherwise. + + :rtype: :class:`google.cloud.spanner_v1.snapshot.Snapshot` + :returns: A Cloud Spanner snapshot object, ready to use. + """ + if not self._snapshot: + self._snapshot = Snapshot(self._session_checkout(), multi_use=True) + return self._snapshot + def _raise_if_closed(self): """Helper to check the connection state before running a query. Raises an exception if this connection is closed. @@ -294,6 +309,8 @@ def commit(self): This method is non-operational in autocommit mode. """ + self._snapshot = None + if self._autocommit: warnings.warn(AUTOCOMMIT_MODE_WARNING, UserWarning, stacklevel=2) return @@ -316,6 +333,8 @@ def rollback(self): This is a no-op if there is no active transaction or if the connection is in autocommit mode. """ + self._snapshot = None + if self._autocommit: warnings.warn(AUTOCOMMIT_MODE_WARNING, UserWarning, stacklevel=2) elif self._transaction: diff --git a/google/cloud/spanner_dbapi/cursor.py b/google/cloud/spanner_dbapi/cursor.py index f66bf1f0a8..afe0b0ab82 100644 --- a/google/cloud/spanner_dbapi/cursor.py +++ b/google/cloud/spanner_dbapi/cursor.py @@ -398,38 +398,48 @@ def setoutputsize(self, size, column=None): """A no-op, raising an error if the cursor or connection is closed.""" self._raise_if_closed() + def _handle_DQL_with_snapshot(self, snapshot, sql, params): + # Reference + # https://googleapis.dev/python/spanner/latest/session-api.html#google.cloud.spanner_v1.session.Session.execute_sql + sql, params = parse_utils.sql_pyformat_args_to_spanner(sql, params) + res = snapshot.execute_sql( + sql, params=params, param_types=get_param_types(params) + ) + if type(res) == int: + self._row_count = res + self._itr = None + else: + # Immediately using: + # iter(response) + # here, because this Spanner API doesn't provide + # easy mechanisms to detect when only a single item + # is returned or many, yet mixing results that + # are for .fetchone() with those that would result in + # many items returns a RuntimeError if .fetchone() is + # invoked and vice versa. + self._result_set = res + # Read the first element so that the StreamedResultSet can + # return the metadata after a DQL statement. See issue #155. + while True: + try: + self._itr = PeekIterator(self._result_set) + break + except Aborted: + self.connection.retry_transaction() + # Unfortunately, Spanner doesn't seem to send back + # information about the number of rows available. + self._row_count = _UNSET_COUNT + def _handle_DQL(self, sql, params): - with self.connection.database.snapshot() as snapshot: - # Reference - # https://googleapis.dev/python/spanner/latest/session-api.html#google.cloud.spanner_v1.session.Session.execute_sql - sql, params = parse_utils.sql_pyformat_args_to_spanner(sql, params) - res = snapshot.execute_sql( - sql, params=params, param_types=get_param_types(params) + if self.connection.read_only and not self.connection.autocommit: + # initiate or use the existing multi-use snapshot + self._handle_DQL_with_snapshot( + self.connection.snapshot_checkout(), sql, params ) - if type(res) == int: - self._row_count = res - self._itr = None - else: - # Immediately using: - # iter(response) - # here, because this Spanner API doesn't provide - # easy mechanisms to detect when only a single item - # is returned or many, yet mixing results that - # are for .fetchone() with those that would result in - # many items returns a RuntimeError if .fetchone() is - # invoked and vice versa. - self._result_set = res - # Read the first element so that the StreamedResultSet can - # return the metadata after a DQL statement. See issue #155. - while True: - try: - self._itr = PeekIterator(self._result_set) - break - except Aborted: - self.connection.retry_transaction() - # Unfortunately, Spanner doesn't seem to send back - # information about the number of rows available. - self._row_count = _UNSET_COUNT + else: + # execute with single-use snapshot + with self.connection.database.snapshot() as snapshot: + self._handle_DQL_with_snapshot(snapshot, sql, params) def __enter__(self): return self diff --git a/tests/unit/spanner_dbapi/test_connection.py b/tests/unit/spanner_dbapi/test_connection.py index c45c99f61d..5dbffe41af 100644 --- a/tests/unit/spanner_dbapi/test_connection.py +++ b/tests/unit/spanner_dbapi/test_connection.py @@ -201,6 +201,27 @@ def test_transaction_checkout(self): connection._autocommit = True self.assertIsNone(connection.transaction_checkout()) + def test_snapshot_checkout(self): + from google.cloud.spanner_dbapi import Connection + + connection = Connection(INSTANCE, DATABASE) + session_checkout = mock.MagicMock(autospec=True) + connection._session_checkout = session_checkout + + snapshot = connection.snapshot_checkout() + session_checkout.assert_called_once() + + self.assertEqual(snapshot, connection.snapshot_checkout()) + + connection.commit() + self.assertIsNone(connection._snapshot) + + connection.snapshot_checkout() + self.assertIsNotNone(connection._snapshot) + + connection.rollback() + self.assertIsNone(connection._snapshot) + @mock.patch("google.cloud.spanner_v1.Client") def test_close(self, mock_client): from google.cloud.spanner_dbapi import connect From 9f1896aa351d845f91a79acb472ace3d3cacb2d1 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Wed, 11 Aug 2021 11:54:17 +0300 Subject: [PATCH 14/27] return the read_only docstring back, erase excess unit test --- google/cloud/spanner_dbapi/connection.py | 8 +++++++ tests/unit/spanner_dbapi/test_connection.py | 25 --------------------- 2 files changed, 8 insertions(+), 25 deletions(-) diff --git a/google/cloud/spanner_dbapi/connection.py b/google/cloud/spanner_dbapi/connection.py index 7c8399c68a..c38c77d82a 100644 --- a/google/cloud/spanner_dbapi/connection.py +++ b/google/cloud/spanner_dbapi/connection.py @@ -58,6 +58,14 @@ class Connection: If True, the connection will use a single use read-only transaction with strong timestamp bound for each new statement, and will immediately see any changes that have been committed by any other transaction. + If autocommit is false, the connection will automatically start a new multi use read-only transaction + with strong timestamp bound when the first statement is executed. This read-only transaction will be + used for all subsequent statements until either commit() or rollback() is called on the connection. The + read-only transaction will read from a consistent snapshot of the database at the time that the + transaction started. This means that the transaction will not see any changes that have been + committed by other transactions since the start of the read-only transaction. Commit or rolling back + the read-only transaction is semantically the same, and only indicates that the read-only transaction + should end a that a new one should be started when the next statement is executed. """ def __init__(self, instance, database, read_only=False): diff --git a/tests/unit/spanner_dbapi/test_connection.py b/tests/unit/spanner_dbapi/test_connection.py index 5dbffe41af..e9f2ebe91a 100644 --- a/tests/unit/spanner_dbapi/test_connection.py +++ b/tests/unit/spanner_dbapi/test_connection.py @@ -121,31 +121,6 @@ def test_read_only_connection(self): 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._transaction = transaction - - 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 From f11ea8c84c65ee6f149d8261c01a0aa416cc368e Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Mon, 16 Aug 2021 12:26:29 +0300 Subject: [PATCH 15/27] erase excess ifs --- google/cloud/spanner_dbapi/connection.py | 17 +++++++---------- 1 file changed, 7 insertions(+), 10 deletions(-) diff --git a/google/cloud/spanner_dbapi/connection.py b/google/cloud/spanner_dbapi/connection.py index c38c77d82a..2c0b30128e 100644 --- a/google/cloud/spanner_dbapi/connection.py +++ b/google/cloud/spanner_dbapi/connection.py @@ -229,16 +229,15 @@ def _rerun_previous_statements(self): self.connection._transaction = None raise Aborted(status.details) - if not self.read_only: - retried_checksum = ResultsChecksum() - retried_checksum.consume_result(res) - retried_checksum.consume_result(status.code) + retried_checksum = ResultsChecksum() + retried_checksum.consume_result(res) + retried_checksum.consume_result(status.code) - _compare_checksums(checksum, retried_checksum) + _compare_checksums(checksum, retried_checksum) else: res_iter, retried_checksum = self.run_statement(statement, retried=True) # executing all the completed statements - if statement != self._statements[-1] and not self.read_only: + if statement != self._statements[-1]: for res in res_iter: retried_checksum.consume_result(res) @@ -250,13 +249,11 @@ def _rerun_previous_statements(self): while len(retried_checksum) < len(statement.checksum): try: res = next(iter(res_iter)) - if not self.read_only: - retried_checksum.consume_result(res) + retried_checksum.consume_result(res) except StopIteration: break - if not self.read_only: - _compare_checksums(statement.checksum, retried_checksum) + _compare_checksums(statement.checksum, retried_checksum) def transaction_checkout(self): """Get a Cloud Spanner transaction. From d807a2d000d7585516dba696b941bd58f53c3089 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Mon, 23 Aug 2021 10:58:27 +0300 Subject: [PATCH 16/27] add additional check into the snapshot_checkout() method --- google/cloud/spanner_dbapi/connection.py | 8 +++++--- tests/unit/spanner_dbapi/test_connection.py | 7 ++++++- 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/google/cloud/spanner_dbapi/connection.py b/google/cloud/spanner_dbapi/connection.py index 2c0b30128e..a1bd0f33d6 100644 --- a/google/cloud/spanner_dbapi/connection.py +++ b/google/cloud/spanner_dbapi/connection.py @@ -282,9 +282,11 @@ def snapshot_checkout(self): :rtype: :class:`google.cloud.spanner_v1.snapshot.Snapshot` :returns: A Cloud Spanner snapshot object, ready to use. """ - if not self._snapshot: - self._snapshot = Snapshot(self._session_checkout(), multi_use=True) - return self._snapshot + if self.read_only and not self.autocommit: + if not self._snapshot: + self._snapshot = Snapshot(self._session_checkout(), multi_use=True) + + return self._snapshot def _raise_if_closed(self): """Helper to check the connection state before running a query. diff --git a/tests/unit/spanner_dbapi/test_connection.py b/tests/unit/spanner_dbapi/test_connection.py index e9f2ebe91a..1f7e5bbc9f 100644 --- a/tests/unit/spanner_dbapi/test_connection.py +++ b/tests/unit/spanner_dbapi/test_connection.py @@ -179,7 +179,9 @@ def test_transaction_checkout(self): def test_snapshot_checkout(self): from google.cloud.spanner_dbapi import Connection - connection = Connection(INSTANCE, DATABASE) + connection = Connection(INSTANCE, DATABASE, read_only=True) + connection.autocommit = False + session_checkout = mock.MagicMock(autospec=True) connection._session_checkout = session_checkout @@ -197,6 +199,9 @@ def test_snapshot_checkout(self): connection.rollback() self.assertIsNone(connection._snapshot) + connection.autocommit = True + self.assertIsNone(connection.snapshot_checkout()) + @mock.patch("google.cloud.spanner_v1.Client") def test_close(self, mock_client): from google.cloud.spanner_dbapi import connect From c61f212efd77f17ad07f33916d7345e6100cea48 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Mon, 13 Sep 2021 11:50:47 +0300 Subject: [PATCH 17/27] add new style system test --- google/cloud/spanner_dbapi/connection.py | 1 + tests/system/test_dbapi.py | 27 ++++++++++++++++++++++++ 2 files changed, 28 insertions(+) diff --git a/google/cloud/spanner_dbapi/connection.py b/google/cloud/spanner_dbapi/connection.py index a1bd0f33d6..13fab27ca5 100644 --- a/google/cloud/spanner_dbapi/connection.py +++ b/google/cloud/spanner_dbapi/connection.py @@ -285,6 +285,7 @@ def snapshot_checkout(self): if self.read_only and not self.autocommit: if not self._snapshot: self._snapshot = Snapshot(self._session_checkout(), multi_use=True) + self._snapshot.begin() return self._snapshot diff --git a/tests/system/test_dbapi.py b/tests/system/test_dbapi.py index 5cc7df677a..e9ea162c05 100644 --- a/tests/system/test_dbapi.py +++ b/tests/system/test_dbapi.py @@ -17,10 +17,13 @@ import pytest +from google.api_core import exceptions from google.cloud import spanner_v1 from google.cloud.spanner_dbapi.connection import Connection +from google.cloud.spanner_dbapi.exceptions import ProgrammingError from . import _helpers + DATABASE_NAME = "dbapi-txn" DDL_STATEMENTS = ( @@ -350,3 +353,27 @@ def test_DDL_commit(shared_instance, dbapi_database): cur.execute("DROP TABLE Singers") conn.commit() + + +def test_read_only(shared_instance, dbapi_database): + """ + Check that connection set to `read_only=True` uses + ReadOnly transactions. + """ + conn = Connection(shared_instance, dbapi_database, read_only=True) + cur = conn.cursor() + + with pytest.raises( + ProgrammingError, + match="400 DML statements can only be performed in a read-write transaction.", + ): + cur.execute( + """ +UPDATE contacts +SET first_name = 'updated-first-name' +WHERE first_name = 'first-name' +""" + ) + + cur.execute("SELECT * FROM contacts") + conn.commit() From 22e5e73d6e214efbc9f9b14e43db8e5f81e99611 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Mon, 13 Sep 2021 11:58:21 +0300 Subject: [PATCH 18/27] don't use error message regexes --- tests/system/test_dbapi.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/tests/system/test_dbapi.py b/tests/system/test_dbapi.py index 6809d8e212..dc6f2e74dd 100644 --- a/tests/system/test_dbapi.py +++ b/tests/system/test_dbapi.py @@ -370,10 +370,7 @@ def test_read_only(shared_instance, dbapi_database): conn = Connection(shared_instance, dbapi_database, read_only=True) cur = conn.cursor() - with pytest.raises( - ProgrammingError, - match="400 DML statements can only be performed in a read-write transaction.", - ): + with pytest.raises(ProgrammingError): cur.execute( """ UPDATE contacts From 1cdccbe33127aa02c8355f5acbb5c8dfee577c93 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Mon, 13 Sep 2021 12:08:54 +0300 Subject: [PATCH 19/27] erase excess import --- tests/system/test_dbapi.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/system/test_dbapi.py b/tests/system/test_dbapi.py index dc6f2e74dd..83e17fae71 100644 --- a/tests/system/test_dbapi.py +++ b/tests/system/test_dbapi.py @@ -17,7 +17,6 @@ import pytest -from google.api_core import exceptions from google.cloud import spanner_v1 from google.cloud.spanner_dbapi.connection import Connection from google.cloud.spanner_dbapi.exceptions import ProgrammingError From ac8c4b2b62cd228d687decda4ec54b32553164e4 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Tue, 14 Sep 2021 11:26:52 +0300 Subject: [PATCH 20/27] refactor --- google/cloud/spanner_dbapi/cursor.py | 54 +++++++++++-------------- tests/unit/spanner_dbapi/test_cursor.py | 40 +----------------- 2 files changed, 26 insertions(+), 68 deletions(-) diff --git a/google/cloud/spanner_dbapi/cursor.py b/google/cloud/spanner_dbapi/cursor.py index afe0b0ab82..c1657e671a 100644 --- a/google/cloud/spanner_dbapi/cursor.py +++ b/google/cloud/spanner_dbapi/cursor.py @@ -334,8 +334,9 @@ def fetchone(self): except StopIteration: return except Aborted: - self.connection.retry_transaction() - return self.fetchone() + if not self.connection.read_only: + self.connection.retry_transaction() + return self.fetchone() def fetchall(self): """Fetch all (remaining) rows of a query result, returning them as @@ -350,8 +351,9 @@ def fetchall(self): self._checksum.consume_result(row) res.append(row) except Aborted: - self.connection.retry_transaction() - return self.fetchall() + if not self.connection.read_only: + self.connection.retry_transaction() + return self.fetchall() return res @@ -381,8 +383,9 @@ def fetchmany(self, size=None): except StopIteration: break except Aborted: - self.connection.retry_transaction() - return self.fetchmany(size) + if not self.connection.read_only: + self.connection.retry_transaction() + return self.fetchmany(size) return items @@ -405,30 +408,21 @@ def _handle_DQL_with_snapshot(self, snapshot, sql, params): res = snapshot.execute_sql( sql, params=params, param_types=get_param_types(params) ) - if type(res) == int: - self._row_count = res - self._itr = None - else: - # Immediately using: - # iter(response) - # here, because this Spanner API doesn't provide - # easy mechanisms to detect when only a single item - # is returned or many, yet mixing results that - # are for .fetchone() with those that would result in - # many items returns a RuntimeError if .fetchone() is - # invoked and vice versa. - self._result_set = res - # Read the first element so that the StreamedResultSet can - # return the metadata after a DQL statement. See issue #155. - while True: - try: - self._itr = PeekIterator(self._result_set) - break - except Aborted: - self.connection.retry_transaction() - # Unfortunately, Spanner doesn't seem to send back - # information about the number of rows available. - self._row_count = _UNSET_COUNT + # Immediately using: + # iter(response) + # here, because this Spanner API doesn't provide + # easy mechanisms to detect when only a single item + # is returned or many, yet mixing results that + # are for .fetchone() with those that would result in + # many items returns a RuntimeError if .fetchone() is + # invoked and vice versa. + self._result_set = res + # Read the first element so that the StreamedResultSet can + # return the metadata after a DQL statement. See issue #155. + self._itr = PeekIterator(self._result_set) + # Unfortunately, Spanner doesn't seem to send back + # information about the number of rows available. + self._row_count = _UNSET_COUNT def _handle_DQL(self, sql, params): if self.connection.read_only and not self.connection.autocommit: diff --git a/tests/unit/spanner_dbapi/test_cursor.py b/tests/unit/spanner_dbapi/test_cursor.py index 07deffd707..940d44e211 100644 --- a/tests/unit/spanner_dbapi/test_cursor.py +++ b/tests/unit/spanner_dbapi/test_cursor.py @@ -707,14 +707,9 @@ def test_handle_dql(self): ) = mock.MagicMock() cursor = self._make_one(connection) - mock_snapshot.execute_sql.return_value = int(0) + mock_snapshot.execute_sql.return_value = ["0"] cursor._handle_DQL("sql", params=None) - self.assertEqual(cursor._row_count, 0) - self.assertIsNone(cursor._itr) - - mock_snapshot.execute_sql.return_value = "0" - cursor._handle_DQL("sql", params=None) - self.assertEqual(cursor._result_set, "0") + self.assertEqual(cursor._result_set, ["0"]) self.assertIsInstance(cursor._itr, utils.PeekIterator) self.assertEqual(cursor._row_count, _UNSET_COUNT) @@ -831,37 +826,6 @@ def test_peek_iterator_aborted(self, mock_client): retry_mock.assert_called_with() - @mock.patch("google.cloud.spanner_v1.Client") - def test_peek_iterator_aborted_autocommit(self, mock_client): - """ - Checking that an Aborted exception is retried in case it happened while - streaming the first element with a PeekIterator in autocommit mode. - """ - from google.api_core.exceptions import Aborted - from google.cloud.spanner_dbapi.connection import connect - - connection = connect("test-instance", "test-database") - - connection.autocommit = True - cursor = connection.cursor() - with mock.patch( - "google.cloud.spanner_dbapi.utils.PeekIterator.__init__", - side_effect=(Aborted("Aborted"), None), - ): - with mock.patch( - "google.cloud.spanner_dbapi.connection.Connection.retry_transaction" - ) as retry_mock: - with mock.patch( - "google.cloud.spanner_dbapi.connection.Connection.run_statement", - return_value=((1, 2, 3), None), - ): - with mock.patch( - "google.cloud.spanner_v1.database.Database.snapshot" - ): - cursor.execute("SELECT * FROM table_name") - - retry_mock.assert_called_with() - @mock.patch("google.cloud.spanner_v1.Client") def test_fetchone_retry_aborted(self, mock_client): """Check that aborted fetch re-executing transaction.""" From 6973748ea2cd7c09ebf4543ef03cb51da3b85210 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Thu, 16 Sep 2021 11:31:11 +0300 Subject: [PATCH 21/27] add unit test to check that read-only transactions are not retried --- tests/unit/spanner_dbapi/test_connection.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/tests/unit/spanner_dbapi/test_connection.py b/tests/unit/spanner_dbapi/test_connection.py index e30626ae02..34e50255f9 100644 --- a/tests/unit/spanner_dbapi/test_connection.py +++ b/tests/unit/spanner_dbapi/test_connection.py @@ -121,6 +121,26 @@ def test_read_only_connection(self): connection.read_only = False self.assertFalse(connection.read_only) + def test_read_only_not_retried(self): + """ + Testing the unlikely case of a read-only transaction + failed with Aborted exception. In this case the + transaction should not be automatically retried. + """ + from google.api_core.exceptions import Aborted + + connection = self._make_connection(read_only=True) + connection.retry_transaction = mock.Mock() + + cursor = connection.cursor() + cursor._itr = mock.Mock(__next__=mock.Mock(side_effect=Aborted("Aborted"),)) + + cursor.fetchone() + cursor.fetchall() + cursor.fetchmany(5) + + connection.retry_transaction.assert_not_called() + @staticmethod def _make_pool(): from google.cloud.spanner_v1.pool import AbstractSessionPool From f18e1022262c06524a1dd8f1345eac5dcbbfe29b Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Tue, 21 Sep 2021 11:37:40 +0300 Subject: [PATCH 22/27] feat(db_api): support stale reads --- google/cloud/spanner_dbapi/connection.py | 47 ++++++- google/cloud/spanner_dbapi/cursor.py | 4 +- tests/system/test_dbapi.py | 31 +++++ tests/unit/spanner_dbapi/test_connection.py | 134 ++++++++++++++++++-- 4 files changed, 202 insertions(+), 14 deletions(-) diff --git a/google/cloud/spanner_dbapi/connection.py b/google/cloud/spanner_dbapi/connection.py index ba9fea3858..71c8590e7e 100644 --- a/google/cloud/spanner_dbapi/connection.py +++ b/google/cloud/spanner_dbapi/connection.py @@ -87,6 +87,7 @@ def __init__(self, instance, database, read_only=False): # connection close self._own_pool = True self._read_only = read_only + self._staleness = None @property def autocommit(self): @@ -165,6 +166,48 @@ def read_only(self, value): ) self._read_only = value + @property + def staleness(self): + """Current read staleness option value of this `Connection`. + + Returns: + dict: Staleness type and value. + """ + return self._staleness + + @staleness.setter + def staleness(self, value): + """Read staleness option setter. + + Args: + value (dict): Staleness type and value. + """ + if self.inside_transaction: + raise ValueError( + "`staleness` option can't be changed while a transaction is in progress. " + "Commit or rollback the current transaction and try again." + ) + + if value is not None: + one_of_expected = False + for opt in ( + "read_timestamp", + "min_read_timestamp", + "max_staleness", + "exact_staleness", + ): + if opt in value and value[opt]: + one_of_expected = True + break + + if not one_of_expected: + raise ValueError( + "Expected one of the following staleness options: " + "read_timestamp, min_read_timestamp, max_staleness, exact_staleness." + ) + + self._staleness = value + def _session_checkout(self): """Get a Cloud Spanner session from the pool. @@ -284,7 +327,9 @@ def snapshot_checkout(self): """ if self.read_only and not self.autocommit: if not self._snapshot: - self._snapshot = Snapshot(self._session_checkout(), multi_use=True) + self._snapshot = Snapshot( + self._session_checkout(), multi_use=True, **self._staleness or {} + ) self._snapshot.begin() return self._snapshot diff --git a/google/cloud/spanner_dbapi/cursor.py b/google/cloud/spanner_dbapi/cursor.py index 64df68b362..3155a729a8 100644 --- a/google/cloud/spanner_dbapi/cursor.py +++ b/google/cloud/spanner_dbapi/cursor.py @@ -433,7 +433,9 @@ def _handle_DQL(self, sql, params): ) else: # execute with single-use snapshot - with self.connection.database.snapshot() as snapshot: + with self.connection.database.snapshot( + **self.connection.staleness or {} + ) as snapshot: self._handle_DQL_with_snapshot(snapshot, sql, params) def __enter__(self): diff --git a/tests/system/test_dbapi.py b/tests/system/test_dbapi.py index 128a72de06..22f949473a 100644 --- a/tests/system/test_dbapi.py +++ b/tests/system/test_dbapi.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import datetime import hashlib import pickle import pkg_resources @@ -19,6 +20,7 @@ import pytest from google.cloud import spanner_v1 +from google.cloud._helpers import UTC from google.cloud.spanner_dbapi.connection import connect from google.cloud.spanner_dbapi.connection import Connection from google.cloud.spanner_dbapi.exceptions import ProgrammingError @@ -391,3 +393,32 @@ def test_read_only(shared_instance, dbapi_database): cur.execute("SELECT * FROM contacts") conn.commit() + + +def test_staleness(shared_instance, dbapi_database): + """Check the DB API `staleness` option.""" + conn = Connection(shared_instance, dbapi_database) + cursor = conn.cursor() + + before_insert = datetime.datetime.utcnow().replace(tzinfo=UTC) + + cursor.execute( + """ +INSERT INTO contacts (contact_id, first_name, last_name, email) +VALUES (1, 'first-name', 'last-name', 'test.email@example.com') + """ + ) + conn.commit() + + conn.read_only = True + conn.staleness = {"read_timestamp": before_insert} + cursor.execute("SELECT * FROM contacts") + conn.commit() + assert len(cursor.fetchall()) == 0 + + conn.staleness = None + cursor.execute("SELECT * FROM contacts") + conn.commit() + assert len(cursor.fetchall()) == 1 + + conn.close() diff --git a/tests/unit/spanner_dbapi/test_connection.py b/tests/unit/spanner_dbapi/test_connection.py index 34e50255f9..0f51c56283 100644 --- a/tests/unit/spanner_dbapi/test_connection.py +++ b/tests/unit/spanner_dbapi/test_connection.py @@ -14,6 +14,7 @@ """Cloud Spanner DB-API Connection class unit tests.""" +import datetime import mock import unittest import warnings @@ -688,9 +689,6 @@ def test_retry_transaction_w_empty_response(self): run_mock.assert_called_with(statement, retried=True) def test_validate_ok(self): - def exit_func(self, exc_type, exc_value, traceback): - pass - connection = self._make_connection() # mock snapshot context manager @@ -699,7 +697,7 @@ def exit_func(self, exc_type, exc_value, traceback): snapshot_ctx = mock.Mock() snapshot_ctx.__enter__ = mock.Mock(return_value=snapshot_obj) - snapshot_ctx.__exit__ = exit_func + snapshot_ctx.__exit__ = exit_ctx_func snapshot_method = mock.Mock(return_value=snapshot_ctx) connection.database.snapshot = snapshot_method @@ -710,9 +708,6 @@ def exit_func(self, exc_type, exc_value, traceback): def test_validate_fail(self): from google.cloud.spanner_dbapi.exceptions import OperationalError - def exit_func(self, exc_type, exc_value, traceback): - pass - connection = self._make_connection() # mock snapshot context manager @@ -721,7 +716,7 @@ def exit_func(self, exc_type, exc_value, traceback): snapshot_ctx = mock.Mock() snapshot_ctx.__enter__ = mock.Mock(return_value=snapshot_obj) - snapshot_ctx.__exit__ = exit_func + snapshot_ctx.__exit__ = exit_ctx_func snapshot_method = mock.Mock(return_value=snapshot_ctx) connection.database.snapshot = snapshot_method @@ -734,9 +729,6 @@ def exit_func(self, exc_type, exc_value, traceback): def test_validate_error(self): from google.cloud.exceptions import NotFound - def exit_func(self, exc_type, exc_value, traceback): - pass - connection = self._make_connection() # mock snapshot context manager @@ -745,7 +737,7 @@ def exit_func(self, exc_type, exc_value, traceback): snapshot_ctx = mock.Mock() snapshot_ctx.__enter__ = mock.Mock(return_value=snapshot_obj) - snapshot_ctx.__exit__ = exit_func + snapshot_ctx.__exit__ = exit_ctx_func snapshot_method = mock.Mock(return_value=snapshot_ctx) connection.database.snapshot = snapshot_method @@ -763,3 +755,121 @@ def test_validate_closed(self): with self.assertRaises(InterfaceError): connection.validate() + + def test_staleness_invalid_value(self): + """Check that `staleness` property accepts only correct values.""" + connection = self._make_connection() + + # incorrect value + with self.assertRaises(ValueError): + connection.staleness = {"read_timestamp": None} + + # incorrect staleness type + with self.assertRaises(ValueError): + connection.staleness = {"something": 4} + + # no expected staleness types + with self.assertRaises(ValueError): + connection.staleness = {} + + def test_staleness_inside_transaction(self): + """ + Check that it's impossible to change the `steleness` + option if a transaction is in progress. + """ + connection = self._make_connection() + connection._transaction = mock.Mock(committed=False, rolled_back=False) + + with self.assertRaises(ValueError): + connection.staleness = {"read_timestamp": datetime.datetime(2021, 9, 21)} + + def test_staleness_multi_use(self): + """ + Check that `steleness` option is correctly + sent to the `Snapshot()` constructor. + + READ_ONLY, NOT AUTOCOMMIT + """ + timestamp = datetime.datetime(2021, 9, 20) + + connection = self._make_connection() + connection._session = "session" + connection.read_only = True + connection.staleness = {"read_timestamp": timestamp} + + with mock.patch( + "google.cloud.spanner_dbapi.connection.Snapshot" + ) as snapshot_mock: + connection.snapshot_checkout() + + snapshot_mock.assert_called_with( + "session", multi_use=True, read_timestamp=timestamp + ) + + def test_staleness_single_use_autocommit(self): + """ + Check that `staleness` option is correctly + sent to the snapshot context manager. + + NOT READ_ONLY, AUTOCOMMIT + """ + timestamp = datetime.datetime(2021, 9, 20) + + connection = self._make_connection() + connection._session_checkout = mock.MagicMock(autospec=True) + + connection.autocommit = True + connection.staleness = {"read_timestamp": timestamp} + + # mock snapshot context manager + snapshot_obj = mock.Mock() + snapshot_obj.execute_sql = mock.Mock(return_value=[1]) + + snapshot_ctx = mock.Mock() + snapshot_ctx.__enter__ = mock.Mock(return_value=snapshot_obj) + snapshot_ctx.__exit__ = exit_ctx_func + snapshot_method = mock.Mock(return_value=snapshot_ctx) + + connection.database.snapshot = snapshot_method + + cursor = connection.cursor() + cursor.execute("SELECT 1") + + connection.database.snapshot.assert_called_with(read_timestamp=timestamp) + + def test_staleness_single_use_readonly_autocommit(self): + """ + Check that `staleness` option is correctly sent to the + snapshot context manager while in `autocommit` mode. + + READ_ONLY, AUTOCOMMIT + """ + timestamp = datetime.datetime(2021, 9, 20) + + connection = self._make_connection() + connection.autocommit = True + connection.read_only = True + connection._session_checkout = mock.MagicMock(autospec=True) + + connection.staleness = {"read_timestamp": timestamp} + + # mock snapshot context manager + snapshot_obj = mock.Mock() + snapshot_obj.execute_sql = mock.Mock(return_value=[1]) + + snapshot_ctx = mock.Mock() + snapshot_ctx.__enter__ = mock.Mock(return_value=snapshot_obj) + snapshot_ctx.__exit__ = exit_ctx_func + snapshot_method = mock.Mock(return_value=snapshot_ctx) + + connection.database.snapshot = snapshot_method + + cursor = connection.cursor() + cursor.execute("SELECT 1") + + connection.database.snapshot.assert_called_with(read_timestamp=timestamp) + + +def exit_ctx_func(self, exc_type, exc_value, traceback): + """Context __exit__ method mock.""" + pass From b22c31e81fd0b23df10a8301f60ff0a907d85368 Mon Sep 17 00:00:00 2001 From: Ilya Gurov Date: Mon, 11 Oct 2021 12:53:12 +0300 Subject: [PATCH 23/27] Apply suggestions from code review Co-authored-by: larkee <31196561+larkee@users.noreply.github.com> --- tests/unit/spanner_dbapi/test_connection.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/unit/spanner_dbapi/test_connection.py b/tests/unit/spanner_dbapi/test_connection.py index 0f51c56283..e6acb78290 100644 --- a/tests/unit/spanner_dbapi/test_connection.py +++ b/tests/unit/spanner_dbapi/test_connection.py @@ -774,7 +774,7 @@ def test_staleness_invalid_value(self): def test_staleness_inside_transaction(self): """ - Check that it's impossible to change the `steleness` + Check that it's impossible to change the `staleness` option if a transaction is in progress. """ connection = self._make_connection() @@ -785,7 +785,7 @@ def test_staleness_inside_transaction(self): def test_staleness_multi_use(self): """ - Check that `steleness` option is correctly + Check that `staleness` option is correctly sent to the `Snapshot()` constructor. READ_ONLY, NOT AUTOCOMMIT From 6a3fedbb50c9ccaab2454a971de8ea00138771da Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Mon, 11 Oct 2021 13:00:23 +0300 Subject: [PATCH 24/27] move `or None` into the property --- google/cloud/spanner_dbapi/connection.py | 4 ++-- google/cloud/spanner_dbapi/cursor.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/google/cloud/spanner_dbapi/connection.py b/google/cloud/spanner_dbapi/connection.py index 71c8590e7e..0c36fe58df 100644 --- a/google/cloud/spanner_dbapi/connection.py +++ b/google/cloud/spanner_dbapi/connection.py @@ -173,7 +173,7 @@ def staleness(self): Returns: dict: Staleness type and value. """ - return self._staleness + return self._staleness or {} @staleness.setter def staleness(self, value): @@ -328,7 +328,7 @@ def snapshot_checkout(self): if self.read_only and not self.autocommit: if not self._snapshot: self._snapshot = Snapshot( - self._session_checkout(), multi_use=True, **self._staleness or {} + self._session_checkout(), multi_use=True, **self.staleness ) self._snapshot.begin() diff --git a/google/cloud/spanner_dbapi/cursor.py b/google/cloud/spanner_dbapi/cursor.py index 3155a729a8..9054cbf268 100644 --- a/google/cloud/spanner_dbapi/cursor.py +++ b/google/cloud/spanner_dbapi/cursor.py @@ -434,7 +434,7 @@ def _handle_DQL(self, sql, params): else: # execute with single-use snapshot with self.connection.database.snapshot( - **self.connection.staleness or {} + **self.connection.staleness ) as snapshot: self._handle_DQL_with_snapshot(snapshot, sql, params) From 6ad2c658b4187b8f1ce493f158156464238b70c8 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Mon, 1 Nov 2021 12:49:10 +0300 Subject: [PATCH 25/27] fix the check --- google/cloud/spanner_dbapi/connection.py | 28 ++++++++------------- tests/unit/spanner_dbapi/test_connection.py | 20 +++++++++------ 2 files changed, 24 insertions(+), 24 deletions(-) diff --git a/google/cloud/spanner_dbapi/connection.py b/google/cloud/spanner_dbapi/connection.py index 0c36fe58df..e6d1d64db1 100644 --- a/google/cloud/spanner_dbapi/connection.py +++ b/google/cloud/spanner_dbapi/connection.py @@ -188,23 +188,17 @@ def staleness(self, value): "Commit or rollback the current transaction and try again." ) - if value is not None: - one_of_expected = False - for opt in ( - "read_timestamp", - "min_read_timestamp", - "max_staleness", - "exact_staleness", - ): - if opt in value and value[opt]: - one_of_expected = True - break - - if not one_of_expected: - raise ValueError( - "Expected one of the following staleness options: " - "read_timestamp, min_read_timestamp, max_staleness, exact_staleness." - ) + possible_opts = ( + "read_timestamp", + "min_read_timestamp", + "max_staleness", + "exact_staleness", + ) + if value is not None and sum([opt in value for opt in possible_opts]) != 1: + raise ValueError( + "Expected one of the following staleness options: " + "read_timestamp, min_read_timestamp, max_staleness, exact_staleness." + ) self._staleness = value diff --git a/tests/unit/spanner_dbapi/test_connection.py b/tests/unit/spanner_dbapi/test_connection.py index e6acb78290..7902de6405 100644 --- a/tests/unit/spanner_dbapi/test_connection.py +++ b/tests/unit/spanner_dbapi/test_connection.py @@ -134,7 +134,11 @@ def test_read_only_not_retried(self): connection.retry_transaction = mock.Mock() cursor = connection.cursor() - cursor._itr = mock.Mock(__next__=mock.Mock(side_effect=Aborted("Aborted"),)) + cursor._itr = mock.Mock( + __next__=mock.Mock( + side_effect=Aborted("Aborted"), + ) + ) cursor.fetchone() cursor.fetchall() @@ -574,7 +578,10 @@ def test_retry_aborted_retry(self, mock_client): connection.retry_transaction() run_mock.assert_has_calls( - (mock.call(statement, retried=True), mock.call(statement, retried=True),) + ( + mock.call(statement, retried=True), + mock.call(statement, retried=True), + ) ) def test_retry_transaction_raise_max_internal_retries(self): @@ -631,7 +638,10 @@ def test_retry_aborted_retry_without_delay(self, mock_client): connection.retry_transaction() run_mock.assert_has_calls( - (mock.call(statement, retried=True), mock.call(statement, retried=True),) + ( + mock.call(statement, retried=True), + mock.call(statement, retried=True), + ) ) def test_retry_transaction_w_multiple_statement(self): @@ -760,10 +770,6 @@ def test_staleness_invalid_value(self): """Check that `staleness` property accepts only correct values.""" connection = self._make_connection() - # incorrect value - with self.assertRaises(ValueError): - connection.staleness = {"read_timestamp": None} - # incorrect staleness type with self.assertRaises(ValueError): connection.staleness = {"something": 4} From 45fdf8e6f10cfba1a071452fc9fe82694cb08268 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Mon, 1 Nov 2021 14:19:17 +0300 Subject: [PATCH 26/27] lint fix --- google/cloud/spanner_dbapi/connection.py | 7 +++++-- tests/unit/spanner_dbapi/test_connection.py | 16 +++------------- 2 files changed, 8 insertions(+), 15 deletions(-) diff --git a/google/cloud/spanner_dbapi/connection.py b/google/cloud/spanner_dbapi/connection.py index e6d1d64db1..3d29c15109 100644 --- a/google/cloud/spanner_dbapi/connection.py +++ b/google/cloud/spanner_dbapi/connection.py @@ -440,7 +440,8 @@ def run_statement(self, statement, retried=False): ) else: _execute_insert_heterogenous( - transaction, parts.get("sql_params_list"), + transaction, + parts.get("sql_params_list"), ) return ( iter(()), @@ -449,7 +450,9 @@ def run_statement(self, statement, retried=False): return ( transaction.execute_sql( - statement.sql, statement.params, param_types=statement.param_types, + statement.sql, + statement.params, + param_types=statement.param_types, ), ResultsChecksum() if retried else statement.checksum, ) diff --git a/tests/unit/spanner_dbapi/test_connection.py b/tests/unit/spanner_dbapi/test_connection.py index 7902de6405..0eea3eaf5b 100644 --- a/tests/unit/spanner_dbapi/test_connection.py +++ b/tests/unit/spanner_dbapi/test_connection.py @@ -134,11 +134,7 @@ def test_read_only_not_retried(self): connection.retry_transaction = mock.Mock() cursor = connection.cursor() - cursor._itr = mock.Mock( - __next__=mock.Mock( - side_effect=Aborted("Aborted"), - ) - ) + cursor._itr = mock.Mock(__next__=mock.Mock(side_effect=Aborted("Aborted"),)) cursor.fetchone() cursor.fetchall() @@ -578,10 +574,7 @@ def test_retry_aborted_retry(self, mock_client): connection.retry_transaction() run_mock.assert_has_calls( - ( - mock.call(statement, retried=True), - mock.call(statement, retried=True), - ) + (mock.call(statement, retried=True), mock.call(statement, retried=True),) ) def test_retry_transaction_raise_max_internal_retries(self): @@ -638,10 +631,7 @@ def test_retry_aborted_retry_without_delay(self, mock_client): connection.retry_transaction() run_mock.assert_has_calls( - ( - mock.call(statement, retried=True), - mock.call(statement, retried=True), - ) + (mock.call(statement, retried=True), mock.call(statement, retried=True),) ) def test_retry_transaction_w_multiple_statement(self): From e50875bdc9a144ef51417d6396794d406e987abf Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Wed, 10 Nov 2021 11:53:38 +0300 Subject: [PATCH 27/27] lint fix --- google/cloud/spanner_dbapi/connection.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/google/cloud/spanner_dbapi/connection.py b/google/cloud/spanner_dbapi/connection.py index 3d29c15109..e6d1d64db1 100644 --- a/google/cloud/spanner_dbapi/connection.py +++ b/google/cloud/spanner_dbapi/connection.py @@ -440,8 +440,7 @@ def run_statement(self, statement, retried=False): ) else: _execute_insert_heterogenous( - transaction, - parts.get("sql_params_list"), + transaction, parts.get("sql_params_list"), ) return ( iter(()), @@ -450,9 +449,7 @@ def run_statement(self, statement, retried=False): return ( transaction.execute_sql( - statement.sql, - statement.params, - param_types=statement.param_types, + statement.sql, statement.params, param_types=statement.param_types, ), ResultsChecksum() if retried else statement.checksum, )