feat(dbapi): use inline begin to eliminate BeginTransaction RPC

google/cloud/spanner_dbapi/connection.py

@@ -392,7 +392,14 @@ def transaction_checkout(self):
         this connection yet. Return the started one otherwise.
 
         This method is a no-op if the connection is in autocommit mode and no
-        explicit transaction has been started
+        explicit transaction has been started.
+
+        The transaction is returned without calling ``begin()``. The
+        underlying ``Transaction.execute_sql`` and ``execute_update``
+        methods detect ``_transaction_id is None`` and use *inline begin*
+        — piggybacking a ``BeginTransaction`` on the first RPC via
+        ``TransactionSelector(begin=...)``. This eliminates a separate
+        ``BeginTransaction`` RPC round-trip per transaction.
 
         :rtype: :class:`google.cloud.spanner_v1.transaction.Transaction`
         :returns: A Cloud Spanner transaction object, ready to use.
@@ -410,7 +417,6 @@ def transaction_checkout(self):
                 self.transaction_tag = None
                 self._snapshot = None
                 self._spanner_transaction_started = True
-                self._transaction.begin()
 
             return self._transaction
 

tests/unit/spanner_dbapi/test_connection.py

@@ -211,6 +211,26 @@ def test_transaction_checkout(self):
         connection._autocommit = True
         self.assertIsNone(connection.transaction_checkout())
 
+    def test_transaction_checkout_does_not_call_begin(self):
+        """transaction_checkout must not call Transaction.begin().
+
+        The transaction should be returned with _transaction_id=None so that
+        execute_sql/execute_update can use inline begin via
+        TransactionSelector(begin=...), eliminating a separate
+        BeginTransaction RPC.
+        """
+        connection = Connection(INSTANCE, DATABASE)
+        mock_session = mock.MagicMock()
+        mock_transaction = mock.MagicMock()
+        mock_session.transaction.return_value = mock_transaction
+        connection._session_checkout = mock.MagicMock(return_value=mock_session)
+
+        txn = connection.transaction_checkout()
+
+        self.assertEqual(txn, mock_transaction)
+        self.assertTrue(connection._spanner_transaction_started)
+        mock_transaction.begin.assert_not_called()
+
     def test_snapshot_checkout(self):
         connection = build_connection(read_only=True)
         connection.autocommit = False