Merge branch 'main' into support-db-roles-in-connect

Author: olavloite

.github/workflows/mock_server_tests.yaml

@@ -5,7 +5,7 @@ on:
   pull_request:
 name: Run Spanner tests against an in-mem mock server
 jobs:
-  system-tests:
+  mock-server-tests:
     runs-on: ubuntu-latest
 
     steps:

.github/workflows/presubmit.yaml

@@ -0,0 +1,42 @@
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+name: Presubmit checks
+permissions:
+  contents: read
+  pull-requests: write
+jobs:
+  lint:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: 3.8
+      - name: Install nox
+        run: python -m pip install nox
+      - name: Check formatting
+        run: nox -s lint
+  units:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python: ["3.8", "3.9", "3.10", "3.11", "3.12", "3.13"]
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{matrix.python}}
+      - name: Install nox
+        run: python -m pip install nox
+      - name: Run unit tests
+        run: nox -s unit-${{matrix.python}}

.kokoro/presubmit/presubmit.cfg

@@ -1,7 +1,7 @@
 # Format: //devtools/kokoro/config/proto/build.proto
 
-# Disable system tests.
+# Only run a subset of all nox sessions
 env_vars: {
-    key: "RUN_SYSTEM_TESTS"
-    value: "false"
+    key: "NOX_SESSION"
+    value: "unit-3.8 unit-3.12 cover docs docfx"
 }

google/cloud/spanner_dbapi/connection.py

@@ -17,6 +17,8 @@
 
 from google.api_core.exceptions import Aborted
 from google.api_core.gapic_v1.client_info import ClientInfo
+from google.auth.credentials import AnonymousCredentials
+
 from google.cloud import spanner_v1 as spanner
 from google.cloud.spanner_dbapi import partition_helper
 from google.cloud.spanner_dbapi.batch_dml_executor import BatchMode, BatchDmlExecutor
@@ -789,11 +791,15 @@ def connect(
                 route_to_leader_enabled=route_to_leader_enabled,
             )
         else:
+            client_options = None
+            if isinstance(credentials, AnonymousCredentials):
+                client_options = kwargs.get("client_options")
             client = spanner.Client(
                 project=project,
                 credentials=credentials,
                 client_info=client_info,
                 route_to_leader_enabled=route_to_leader_enabled,
+                client_options=client_options,
             )
     else:
         if project is not None and client.project != project:

google/cloud/spanner_dbapi/transaction_helper.py

@@ -162,7 +162,7 @@ def add_execute_statement_for_retry(
         self._last_statement_details_per_cursor[cursor] = last_statement_result_details
         self._statement_result_details_list.append(last_statement_result_details)
 
-    def retry_transaction(self):
+    def retry_transaction(self, default_retry_delay=None):
         """Retry the aborted transaction.
 
         All the statements executed in the original transaction
@@ -202,7 +202,9 @@ def retry_transaction(self):
                             raise RetryAborted(RETRY_ABORTED_ERROR, ex)
                 return
             except Aborted as ex:
-                delay = _get_retry_delay(ex.errors[0], attempt)
+                delay = _get_retry_delay(
+                    ex.errors[0], attempt, default_retry_delay=default_retry_delay
+                )
                 if delay:
                     time.sleep(delay)
 

google/cloud/spanner_v1/_helpers.py

@@ -510,6 +510,7 @@ def _metadata_with_prefix(prefix, **kw):
 def _retry_on_aborted_exception(
     func,
     deadline,
+    default_retry_delay=None,
 ):
     """
     Handles retry logic for Aborted exceptions, considering the deadline.
@@ -520,7 +521,12 @@ def _retry_on_aborted_exception(
             attempts += 1
             return func()
         except Aborted as exc:
-            _delay_until_retry(exc, deadline=deadline, attempts=attempts)
+            _delay_until_retry(
+                exc,
+                deadline=deadline,
+                attempts=attempts,
+                default_retry_delay=default_retry_delay,
+            )
             continue
 
 
@@ -608,7 +614,7 @@ def _metadata_with_span_context(metadata: List[Tuple[str, str]], **kw) -> None:
         inject(setter=OpenTelemetryContextSetter(), carrier=metadata)
 
 
-def _delay_until_retry(exc, deadline, attempts):
+def _delay_until_retry(exc, deadline, attempts, default_retry_delay=None):
     """Helper for :meth:`Session.run_in_transaction`.
 
     Detect retryable abort, and impose server-supplied delay.
@@ -628,15 +634,15 @@ def _delay_until_retry(exc, deadline, attempts):
     if now >= deadline:
         raise
 
-    delay = _get_retry_delay(cause, attempts)
+    delay = _get_retry_delay(cause, attempts, default_retry_delay=default_retry_delay)
     if delay is not None:
         if now + delay > deadline:
             raise
 
         time.sleep(delay)
 
 
-def _get_retry_delay(cause, attempts):
+def _get_retry_delay(cause, attempts, default_retry_delay=None):
     """Helper for :func:`_delay_until_retry`.
 
     :type exc: :class:`grpc.Call`
@@ -658,6 +664,8 @@ def _get_retry_delay(cause, attempts):
         retry_info.ParseFromString(retry_info_pb)
         nanos = retry_info.retry_delay.nanos
         return retry_info.retry_delay.seconds + nanos / 1.0e9
+    if default_retry_delay is not None:
+        return default_retry_delay
 
     return 2**attempts + random.random()
 
@@ -699,6 +707,10 @@ def __radd__(self, n):
         """
         return self.__add__(n)
 
+    def reset(self):
+        with self.__lock:
+            self.__value = 0
+
 
 def _metadata_with_request_id(*args, **kwargs):
     return with_request_id(*args, **kwargs)

google/cloud/spanner_v1/batch.py

@@ -257,9 +257,11 @@ def commit(
             deadline = time.time() + kwargs.get(
                 "timeout_secs", DEFAULT_RETRY_TIMEOUT_SECS
             )
+            default_retry_delay = kwargs.get("default_retry_delay", None)
             response = _retry_on_aborted_exception(
                 method,
                 deadline=deadline,
+                default_retry_delay=default_retry_delay,
             )
         self.committed = response.commit_timestamp
         self.commit_stats = response.commit_stats

google/cloud/spanner_v1/client.py

@@ -70,6 +70,7 @@
 except ImportError:  # pragma: NO COVER
     HAS_GOOGLE_CLOUD_MONITORING_INSTALLED = False
 
+from google.cloud.spanner_v1._helpers import AtomicCounter
 
 _CLIENT_INFO = client_info.ClientInfo(client_library_version=__version__)
 EMULATOR_ENV_VAR = "SPANNER_EMULATOR_HOST"
@@ -182,6 +183,8 @@ class Client(ClientWithProject):
     SCOPE = (SPANNER_ADMIN_SCOPE,)
     """The scopes required for Google Cloud Spanner."""
 
+    NTH_CLIENT = AtomicCounter()
+
     def __init__(
         self,
         project=None,
@@ -241,7 +244,9 @@ def __init__(
                 meter_provider = MeterProvider(
                     metric_readers=[
                         PeriodicExportingMetricReader(
-                            CloudMonitoringMetricsExporter(),
+                            CloudMonitoringMetricsExporter(
+                                project_id=project, credentials=credentials
+                            ),
                             export_interval_millis=METRIC_EXPORT_INTERVAL_MS,
                         )
                     ]
@@ -261,6 +266,12 @@ def __init__(
                 "default_transaction_options must be an instance of DefaultTransactionOptions"
             )
         self._default_transaction_options = default_transaction_options
+        self._nth_client_id = Client.NTH_CLIENT.increment()
+        self._nth_request = AtomicCounter(0)
+
+    @property
+    def _next_nth_request(self):
+        return self._nth_request.increment()
 
     @property
     def credentials(self):

google/cloud/spanner_v1/metrics/metrics_exporter.py

@@ -26,6 +26,7 @@
 from typing import Optional, List, Union, NoReturn, Tuple, Dict
 
 import google.auth
+from google.auth import credentials as ga_credentials
 from google.api.distribution_pb2 import (  # pylint: disable=no-name-in-module
     Distribution,
 )
@@ -111,6 +112,7 @@ def __init__(
         self,
         project_id: Optional[str] = None,
         client: Optional["MetricServiceClient"] = None,
+        credentials: Optional[ga_credentials.Credentials] = None,
     ):
         """Initialize a custom exporter to send metrics for the Spanner Service Metrics."""
         # Default preferred_temporality is all CUMULATIVE so need to customize
@@ -121,6 +123,7 @@ def __init__(
             transport=MetricServiceGrpcTransport(
                 channel=MetricServiceGrpcTransport.create_channel(
                     options=_OPTIONS,
+                    credentials=credentials,
                 )
             )
         )

google/cloud/spanner_v1/request_id_header.py

@@ -37,6 +37,6 @@ def generate_rand_uint64():
 
 def with_request_id(client_id, channel_id, nth_request, attempt, other_metadata=[]):
     req_id = f"{REQ_ID_VERSION}.{REQ_RAND_PROCESS_ID}.{client_id}.{channel_id}.{nth_request}.{attempt}"
-    all_metadata = other_metadata.copy()
+    all_metadata = (other_metadata or []).copy()
     all_metadata.append((REQ_ID_HEADER_KEY, req_id))
     return all_metadata