feat: snapshot isolation (#1318)

* feat: snapshot isolation

* test and refactoring

* tests

* 🦉 Updates from OwlBot post-processor

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

* review comments

* review comments and tests

* lint

* 🦉 Updates from OwlBot post-processor

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

* dataclass for default transaction options

* 🦉 Updates from OwlBot post-processor

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

* 🦉 Updates from OwlBot post-processor

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

* proto merge for transaction options

* failed test cases

* review comments

---------

Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>

Author: surbhigarg92

google/cloud/spanner_v1/__init__.py

@@ -64,7 +64,7 @@
 from .types.type import TypeAnnotationCode
 from .types.type import TypeCode
 from .data_types import JsonObject
-from .transaction import BatchTransactionId
+from .transaction import BatchTransactionId, DefaultTransactionOptions
 
 from google.cloud.spanner_v1 import param_types
 from google.cloud.spanner_v1.client import Client
@@ -149,4 +149,5 @@
     "SpannerClient",
     "SpannerAsyncClient",
     "BatchTransactionId",
+    "DefaultTransactionOptions",
 )

google/cloud/spanner_v1/_helpers.py

@@ -32,6 +32,7 @@
 from google.cloud.spanner_v1 import TypeCode
 from google.cloud.spanner_v1 import ExecuteSqlRequest
 from google.cloud.spanner_v1 import JsonObject
+from google.cloud.spanner_v1 import TransactionOptions
 from google.cloud.spanner_v1.request_id_header import with_request_id
 from google.rpc.error_details_pb2 import RetryInfo
 
@@ -690,3 +691,38 @@ def __radd__(self, n):
 
 def _metadata_with_request_id(*args, **kwargs):
     return with_request_id(*args, **kwargs)
+
+
+def _merge_Transaction_Options(
+    defaultTransactionOptions: TransactionOptions,
+    mergeTransactionOptions: TransactionOptions,
+) -> TransactionOptions:
+    """Merges two TransactionOptions objects.
+
+    - Values from `mergeTransactionOptions` take precedence if set.
+    - Values from `defaultTransactionOptions` are used only if missing.
+
+    Args:
+        defaultTransactionOptions (TransactionOptions): The default transaction options (fallback values).
+        mergeTransactionOptions (TransactionOptions): The main transaction options (overrides when set).
+
+    Returns:
+        TransactionOptions: A merged TransactionOptions object.
+    """
+
+    if defaultTransactionOptions is None:
+        return mergeTransactionOptions
+
+    if mergeTransactionOptions is None:
+        return defaultTransactionOptions
+
+    merged_pb = TransactionOptions()._pb  # Create a new protobuf object
+
+    # Merge defaultTransactionOptions first
+    merged_pb.MergeFrom(defaultTransactionOptions._pb)
+
+    # Merge transactionOptions, ensuring it overrides default values
+    merged_pb.MergeFrom(mergeTransactionOptions._pb)
+
+    # Convert protobuf object back into a TransactionOptions instance
+    return TransactionOptions(merged_pb)

google/cloud/spanner_v1/batch.py

@@ -25,6 +25,7 @@
 from google.cloud.spanner_v1._helpers import (
     _metadata_with_prefix,
     _metadata_with_leader_aware_routing,
+    _merge_Transaction_Options,
 )
 from google.cloud.spanner_v1._opentelemetry_tracing import trace_call
 from google.cloud.spanner_v1 import RequestOptions
@@ -167,6 +168,7 @@ def commit(
         request_options=None,
         max_commit_delay=None,
         exclude_txn_from_change_streams=False,
+        isolation_level=TransactionOptions.IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED,
         **kwargs,
     ):
         """Commit mutations to the database.
@@ -187,6 +189,18 @@ def commit(
                 (Optional) The amount of latency this request is willing to incur
                 in order to improve throughput.
 
+        :type exclude_txn_from_change_streams: bool
+        :param exclude_txn_from_change_streams:
+          (Optional) If true, instructs the transaction to be excluded from being recorded in change streams
+          with the DDL option `allow_txn_exclusion=true`. This does not exclude the transaction from
+          being recorded in the change streams with the DDL option `allow_txn_exclusion` being false or
+          unset.
+
+        :type isolation_level:
+            :class:`google.cloud.spanner_v1.types.TransactionOptions.IsolationLevel`
+        :param isolation_level:
+                (Optional) Sets isolation level for the transaction.
+
         :rtype: datetime
         :returns: timestamp of the committed changes.
         """
@@ -201,6 +215,12 @@ def commit(
         txn_options = TransactionOptions(
             read_write=TransactionOptions.ReadWrite(),
             exclude_txn_from_change_streams=exclude_txn_from_change_streams,
+            isolation_level=isolation_level,
+        )
+
+        txn_options = _merge_Transaction_Options(
+            database.default_transaction_options.default_read_write_transaction_options,
+            txn_options,
         )
         trace_attributes = {"num_mutations": len(self._mutations)}
 

google/cloud/spanner_v1/client.py

@@ -31,6 +31,7 @@
 from google.auth.credentials import AnonymousCredentials
 import google.api_core.client_options
 from google.cloud.client import ClientWithProject
+from typing import Optional
 
 
 from google.cloud.spanner_admin_database_v1 import DatabaseAdminClient
@@ -45,6 +46,7 @@
 from google.cloud.spanner_admin_instance_v1 import ListInstancesRequest
 from google.cloud.spanner_v1 import __version__
 from google.cloud.spanner_v1 import ExecuteSqlRequest
+from google.cloud.spanner_v1 import DefaultTransactionOptions
 from google.cloud.spanner_v1._helpers import _merge_query_options
 from google.cloud.spanner_v1._helpers import _metadata_with_prefix
 from google.cloud.spanner_v1.instance import Instance
@@ -165,6 +167,10 @@ class Client(ClientWithProject):
            or you can use the environment variable `SPANNER_ENABLE_END_TO_END_TRACING=<boolean>`
            to control it.
 
+    :type default_transaction_options: :class:`~google.cloud.spanner_v1.DefaultTransactionOptions`
+        or :class:`dict`
+    :param default_transaction_options: (Optional) Default options to use for all transactions.
+
     :raises: :class:`ValueError <exceptions.ValueError>` if both ``read_only``
              and ``admin`` are :data:`True`
     """
@@ -186,6 +192,7 @@ def __init__(
         route_to_leader_enabled=True,
         directed_read_options=None,
         observability_options=None,
+        default_transaction_options: Optional[DefaultTransactionOptions] = None,
     ):
         self._emulator_host = _get_spanner_emulator_host()
 
@@ -247,6 +254,13 @@ def __init__(
         self._route_to_leader_enabled = route_to_leader_enabled
         self._directed_read_options = directed_read_options
         self._observability_options = observability_options
+        if default_transaction_options is None:
+            default_transaction_options = DefaultTransactionOptions()
+        elif not isinstance(default_transaction_options, DefaultTransactionOptions):
+            raise TypeError(
+                "default_transaction_options must be an instance of DefaultTransactionOptions"
+            )
+        self._default_transaction_options = default_transaction_options
 
     @property
     def credentials(self):
@@ -337,6 +351,17 @@ def observability_options(self):
         """
         return self._observability_options
 
+    @property
+    def default_transaction_options(self):
+        """Getter for default_transaction_options.
+
+        :rtype:
+            :class:`~google.cloud.spanner_v1.DefaultTransactionOptions`
+            or :class:`dict`
+        :returns: The default transaction options that are used by this client for all transactions.
+        """
+        return self._default_transaction_options
+
     @property
     def directed_read_options(self):
         """Getter for directed_read_options.
@@ -482,3 +507,21 @@ def directed_read_options(self, directed_read_options):
             or regions should be used for non-transactional reads or queries.
         """
         self._directed_read_options = directed_read_options
+
+    @default_transaction_options.setter
+    def default_transaction_options(
+        self, default_transaction_options: DefaultTransactionOptions
+    ):
+        """Sets default_transaction_options for the client
+        :type default_transaction_options: :class:`~google.cloud.spanner_v1.DefaultTransactionOptions`
+            or :class:`dict`
+        :param default_transaction_options: Default options to use for transactions.
+        """
+        if default_transaction_options is None:
+            default_transaction_options = DefaultTransactionOptions()
+        elif not isinstance(default_transaction_options, DefaultTransactionOptions):
+            raise TypeError(
+                "default_transaction_options must be an instance of DefaultTransactionOptions"
+            )
+
+        self._default_transaction_options = default_transaction_options

google/cloud/spanner_v1/database.py

@@ -46,6 +46,7 @@
 from google.cloud.spanner_v1 import TypeCode
 from google.cloud.spanner_v1 import TransactionSelector
 from google.cloud.spanner_v1 import TransactionOptions
+from google.cloud.spanner_v1 import DefaultTransactionOptions
 from google.cloud.spanner_v1 import RequestOptions
 from google.cloud.spanner_v1 import SpannerClient
 from google.cloud.spanner_v1._helpers import _merge_query_options
@@ -183,6 +184,9 @@ def __init__(
         self._enable_drop_protection = enable_drop_protection
         self._reconciling = False
         self._directed_read_options = self._instance._client.directed_read_options
+        self.default_transaction_options: DefaultTransactionOptions = (
+            self._instance._client.default_transaction_options
+        )
         self._proto_descriptors = proto_descriptors
 
         if pool is None:
@@ -782,6 +786,7 @@ def batch(
         request_options=None,
         max_commit_delay=None,
         exclude_txn_from_change_streams=False,
+        isolation_level=TransactionOptions.IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED,
         **kw,
     ):
         """Return an object which wraps a batch.
@@ -809,14 +814,21 @@ def batch(
           being recorded in the change streams with the DDL option `allow_txn_exclusion` being false or
           unset.
 
+        :type isolation_level:
+            :class:`google.cloud.spanner_v1.types.TransactionOptions.IsolationLevel`
+        :param isolation_level:
+                (Optional) Sets the isolation level for this transaction. This overrides any default isolation level set for the client.
+
         :rtype: :class:`~google.cloud.spanner_v1.database.BatchCheckout`
         :returns: new wrapper
         """
+
         return BatchCheckout(
             self,
             request_options,
             max_commit_delay,
             exclude_txn_from_change_streams,
+            isolation_level,
             **kw,
         )
 
@@ -888,6 +900,7 @@ def run_in_transaction(self, func, *args, **kw):
                    from being recorded in change streams with the DDL option `allow_txn_exclusion=true`.
                    This does not exclude the transaction from being recorded in the change streams with
                    the DDL option `allow_txn_exclusion` being false or unset.
+                   "isolation_level" sets the isolation level for the transaction.
 
         :rtype: Any
         :returns: The return value of ``func``.
@@ -1178,6 +1191,7 @@ def __init__(
         request_options=None,
         max_commit_delay=None,
         exclude_txn_from_change_streams=False,
+        isolation_level=TransactionOptions.IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED,
         **kw,
     ):
         self._database = database
@@ -1190,6 +1204,7 @@ def __init__(
             self._request_options = request_options
         self._max_commit_delay = max_commit_delay
         self._exclude_txn_from_change_streams = exclude_txn_from_change_streams
+        self._isolation_level = isolation_level
         self._kw = kw
 
     def __enter__(self):
@@ -1211,6 +1226,7 @@ def __exit__(self, exc_type, exc_val, exc_tb):
                     request_options=self._request_options,
                     max_commit_delay=self._max_commit_delay,
                     exclude_txn_from_change_streams=self._exclude_txn_from_change_streams,
+                    isolation_level=self._isolation_level,
                     **self._kw,
                 )
         finally:

google/cloud/spanner_v1/session.py

@@ -39,7 +39,6 @@
 from google.cloud.spanner_v1.batch import Batch
 from google.cloud.spanner_v1.snapshot import Snapshot
 from google.cloud.spanner_v1.transaction import Transaction
-
 from google.cloud.spanner_v1.metrics.metrics_capture import MetricsCapture
 
 
@@ -453,6 +452,7 @@ def run_in_transaction(self, func, *args, **kw):
                    from being recorded in change streams with the DDL option `allow_txn_exclusion=true`.
                    This does not exclude the transaction from being recorded in the change streams with
                    the DDL option `allow_txn_exclusion` being false or unset.
+                   "isolation_level" sets the isolation level for the transaction.
 
         :rtype: Any
         :returns: The return value of ``func``.
@@ -467,6 +467,8 @@ def run_in_transaction(self, func, *args, **kw):
         exclude_txn_from_change_streams = kw.pop(
             "exclude_txn_from_change_streams", None
         )
+        isolation_level = kw.pop("isolation_level", None)
+
         attempts = 0
 
         observability_options = getattr(self._database, "observability_options", None)
@@ -482,6 +484,7 @@ def run_in_transaction(self, func, *args, **kw):
                     txn.exclude_txn_from_change_streams = (
                         exclude_txn_from_change_streams
                     )
+                    txn.isolation_level = isolation_level
                 else:
                     txn = self._transaction