Skip to content
This repository was archived by the owner on Mar 31, 2026. It is now read-only.

Commit cd3b950

Browse files
author
Ilya Gurov
authored
feat(db_api): add an ability to set ReadOnly/ReadWrite connection mode (#475)
1 parent aaec1db commit cd3b950

File tree

5 files changed

+205
-82
lines changed

5 files changed

+205
-82
lines changed

‎google/cloud/spanner_dbapi/connection.py‎

Lines changed: 69 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
from google.api_core.gapic_v1.client_info import ClientInfo
2222
from google.cloud import spanner_v1 as spanner
2323
from google.cloud.spanner_v1.session import _get_retry_delay
24+
from google.cloud.spanner_v1.snapshot import Snapshot
2425

2526
from google.cloud.spanner_dbapi._helpers import _execute_insert_heterogenous
2627
from google.cloud.spanner_dbapi._helpers import _execute_insert_homogenous
@@ -50,15 +51,31 @@ class Connection:
5051
5152
:type database: :class:`~google.cloud.spanner_v1.database.Database`
5253
:param database: The database to which the connection is linked.
54+
55+
:type read_only: bool
56+
:param read_only:
57+
Flag to indicate that the connection may only execute queries and no update or DDL statements.
58+
If True, the connection will use a single use read-only transaction with strong timestamp
59+
bound for each new statement, and will immediately see any changes that have been committed by
60+
any other transaction.
61+
If autocommit is false, the connection will automatically start a new multi use read-only transaction
62+
with strong timestamp bound when the first statement is executed. This read-only transaction will be
63+
used for all subsequent statements until either commit() or rollback() is called on the connection. The
64+
read-only transaction will read from a consistent snapshot of the database at the time that the
65+
transaction started. This means that the transaction will not see any changes that have been
66+
committed by other transactions since the start of the read-only transaction. Commit or rolling back
67+
the read-only transaction is semantically the same, and only indicates that the read-only transaction
68+
should end a that a new one should be started when the next statement is executed.
5369
"""
5470

55-
def __init__(self, instance, database):
71+
def __init__(self, instance, database, read_only=False):
5672
self._instance = instance
5773
self._database = database
5874
self._ddl_statements = []
5975

6076
self._transaction = None
6177
self._session = None
78+
self._snapshot = None
6279
# SQL statements, which were executed
6380
# within the current transaction
6481
self._statements = []
@@ -69,6 +86,7 @@ def __init__(self, instance, database):
6986
# this connection should be cleared on the
7087
# connection close
7188
self._own_pool = True
89+
self._read_only = read_only
7290

7391
@property
7492
def autocommit(self):
@@ -123,6 +141,30 @@ def instance(self):
123141
"""
124142
return self._instance
125143

144+
@property
145+
def read_only(self):
146+
"""Flag: the connection can be used only for database reads.
147+
148+
Returns:
149+
bool:
150+
True if the connection may only be used for database reads.
151+
"""
152+
return self._read_only
153+
154+
@read_only.setter
155+
def read_only(self, value):
156+
"""`read_only` flag setter.
157+
158+
Args:
159+
value (bool): True for ReadOnly mode, False for ReadWrite.
160+
"""
161+
if self.inside_transaction:
162+
raise ValueError(
163+
"Connection read/write mode can't be changed while a transaction is in progress. "
164+
"Commit or rollback the current transaction and try again."
165+
)
166+
self._read_only = value
167+
126168
def _session_checkout(self):
127169
"""Get a Cloud Spanner session from the pool.
128170
@@ -231,6 +273,22 @@ def transaction_checkout(self):
231273

232274
return self._transaction
233275

276+
def snapshot_checkout(self):
277+
"""Get a Cloud Spanner snapshot.
278+
279+
Initiate a new multi-use snapshot, if there is no snapshot in
280+
this connection yet. Return the existing one otherwise.
281+
282+
:rtype: :class:`google.cloud.spanner_v1.snapshot.Snapshot`
283+
:returns: A Cloud Spanner snapshot object, ready to use.
284+
"""
285+
if self.read_only and not self.autocommit:
286+
if not self._snapshot:
287+
self._snapshot = Snapshot(self._session_checkout(), multi_use=True)
288+
self._snapshot.begin()
289+
290+
return self._snapshot
291+
234292
def _raise_if_closed(self):
235293
"""Helper to check the connection state before running a query.
236294
Raises an exception if this connection is closed.
@@ -259,14 +317,18 @@ def commit(self):
259317
260318
This method is non-operational in autocommit mode.
261319
"""
320+
self._snapshot = None
321+
262322
if self._autocommit:
263323
warnings.warn(AUTOCOMMIT_MODE_WARNING, UserWarning, stacklevel=2)
264324
return
265325

266326
self.run_prior_DDL_statements()
267327
if self.inside_transaction:
268328
try:
269-
self._transaction.commit()
329+
if not self.read_only:
330+
self._transaction.commit()
331+
270332
self._release_session()
271333
self._statements = []
272334
except Aborted:
@@ -279,10 +341,14 @@ def rollback(self):
279341
This is a no-op if there is no active transaction or if the connection
280342
is in autocommit mode.
281343
"""
344+
self._snapshot = None
345+
282346
if self._autocommit:
283347
warnings.warn(AUTOCOMMIT_MODE_WARNING, UserWarning, stacklevel=2)
284348
elif self._transaction:
285-
self._transaction.rollback()
349+
if not self.read_only:
350+
self._transaction.rollback()
351+
286352
self._release_session()
287353
self._statements = []
288354

‎google/cloud/spanner_dbapi/cursor.py‎

Lines changed: 47 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -186,6 +186,10 @@ def execute(self, sql, args=None):
186186

187187
# Classify whether this is a read-only SQL statement.
188188
try:
189+
if self.connection.read_only:
190+
self._handle_DQL(sql, args or None)
191+
return
192+
189193
classification = parse_utils.classify_stmt(sql)
190194
if classification == parse_utils.STMT_DDL:
191195
ddl_statements = []
@@ -325,14 +329,15 @@ def fetchone(self):
325329

326330
try:
327331
res = next(self)
328-
if not self.connection.autocommit:
332+
if not self.connection.autocommit and not self.connection.read_only:
329333
self._checksum.consume_result(res)
330334
return res
331335
except StopIteration:
332336
return
333337
except Aborted:
334-
self.connection.retry_transaction()
335-
return self.fetchone()
338+
if not self.connection.read_only:
339+
self.connection.retry_transaction()
340+
return self.fetchone()
336341

337342
def fetchall(self):
338343
"""Fetch all (remaining) rows of a query result, returning them as
@@ -343,12 +348,13 @@ def fetchall(self):
343348
res = []
344349
try:
345350
for row in self:
346-
if not self.connection.autocommit:
351+
if not self.connection.autocommit and not self.connection.read_only:
347352
self._checksum.consume_result(row)
348353
res.append(row)
349354
except Aborted:
350-
self.connection.retry_transaction()
351-
return self.fetchall()
355+
if not self.connection.read_only:
356+
self.connection.retry_transaction()
357+
return self.fetchall()
352358

353359
return res
354360

@@ -372,14 +378,15 @@ def fetchmany(self, size=None):
372378
for i in range(size):
373379
try:
374380
res = next(self)
375-
if not self.connection.autocommit:
381+
if not self.connection.autocommit and not self.connection.read_only:
376382
self._checksum.consume_result(res)
377383
items.append(res)
378384
except StopIteration:
379385
break
380386
except Aborted:
381-
self.connection.retry_transaction()
382-
return self.fetchmany(size)
387+
if not self.connection.read_only:
388+
self.connection.retry_transaction()
389+
return self.fetchmany(size)
383390

384391
return items
385392

@@ -395,38 +402,39 @@ def setoutputsize(self, size, column=None):
395402
"""A no-op, raising an error if the cursor or connection is closed."""
396403
self._raise_if_closed()
397404

405+
def _handle_DQL_with_snapshot(self, snapshot, sql, params):
406+
# Reference
407+
# https://googleapis.dev/python/spanner/latest/session-api.html#google.cloud.spanner_v1.session.Session.execute_sql
408+
sql, params = parse_utils.sql_pyformat_args_to_spanner(sql, params)
409+
res = snapshot.execute_sql(
410+
sql, params=params, param_types=get_param_types(params)
411+
)
412+
# Immediately using:
413+
# iter(response)
414+
# here, because this Spanner API doesn't provide
415+
# easy mechanisms to detect when only a single item
416+
# is returned or many, yet mixing results that
417+
# are for .fetchone() with those that would result in
418+
# many items returns a RuntimeError if .fetchone() is
419+
# invoked and vice versa.
420+
self._result_set = res
421+
# Read the first element so that the StreamedResultSet can
422+
# return the metadata after a DQL statement. See issue #155.
423+
self._itr = PeekIterator(self._result_set)
424+
# Unfortunately, Spanner doesn't seem to send back
425+
# information about the number of rows available.
426+
self._row_count = _UNSET_COUNT
427+
398428
def _handle_DQL(self, sql, params):
399-
with self.connection.database.snapshot() as snapshot:
400-
# Reference
401-
# https://googleapis.dev/python/spanner/latest/session-api.html#google.cloud.spanner_v1.session.Session.execute_sql
402-
sql, params = parse_utils.sql_pyformat_args_to_spanner(sql, params)
403-
res = snapshot.execute_sql(
404-
sql, params=params, param_types=get_param_types(params)
429+
if self.connection.read_only and not self.connection.autocommit:
430+
# initiate or use the existing multi-use snapshot
431+
self._handle_DQL_with_snapshot(
432+
self.connection.snapshot_checkout(), sql, params
405433
)
406-
if type(res) == int:
407-
self._row_count = res
408-
self._itr = None
409-
else:
410-
# Immediately using:
411-
# iter(response)
412-
# here, because this Spanner API doesn't provide
413-
# easy mechanisms to detect when only a single item
414-
# is returned or many, yet mixing results that
415-
# are for .fetchone() with those that would result in
416-
# many items returns a RuntimeError if .fetchone() is
417-
# invoked and vice versa.
418-
self._result_set = res
419-
# Read the first element so that the StreamedResultSet can
420-
# return the metadata after a DQL statement. See issue #155.
421-
while True:
422-
try:
423-
self._itr = PeekIterator(self._result_set)
424-
break
425-
except Aborted:
426-
self.connection.retry_transaction()
427-
# Unfortunately, Spanner doesn't seem to send back
428-
# information about the number of rows available.
429-
self._row_count = _UNSET_COUNT
434+
else:
435+
# execute with single-use snapshot
436+
with self.connection.database.snapshot() as snapshot:
437+
self._handle_DQL_with_snapshot(snapshot, sql, params)
430438

431439
def __enter__(self):
432440
return self

‎tests/system/test_dbapi.py‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,9 +19,11 @@
1919

2020
from google.cloud import spanner_v1
2121
from google.cloud.spanner_dbapi.connection import connect, Connection
22+
from google.cloud.spanner_dbapi.exceptions import ProgrammingError
2223
from google.cloud.spanner_v1 import JsonObject
2324
from . import _helpers
2425

26+
2527
DATABASE_NAME = "dbapi-txn"
2628

2729
DDL_STATEMENTS = (
@@ -406,3 +408,24 @@ def test_user_agent(shared_instance, dbapi_database):
406408
conn.instance._client._client_info.user_agent
407409
== "dbapi/" + pkg_resources.get_distribution("google-cloud-spanner").version
408410
)
411+
412+
413+
def test_read_only(shared_instance, dbapi_database):
414+
"""
415+
Check that connection set to `read_only=True` uses
416+
ReadOnly transactions.
417+
"""
418+
conn = Connection(shared_instance, dbapi_database, read_only=True)
419+
cur = conn.cursor()
420+
421+
with pytest.raises(ProgrammingError):
422+
cur.execute(
423+
"""
424+
UPDATE contacts
425+
SET first_name = 'updated-first-name'
426+
WHERE first_name = 'first-name'
427+
"""
428+
)
429+
430+
cur.execute("SELECT * FROM contacts")
431+
conn.commit()

0 commit comments

Comments
 (0)