diff --git a/gcloud/datastore/api.py b/gcloud/datastore/api.py
index a63c5bddf088..329cb1e5f10e 100644
--- a/gcloud/datastore/api.py
+++ b/gcloud/datastore/api.py
@@ -19,8 +19,8 @@
 """
 
 from gcloud.datastore import _implicit_environ
-from gcloud.datastore.batch import _BATCHES
 from gcloud.datastore.batch import Batch
+from gcloud.datastore.transaction import Transaction
 from gcloud.datastore import helpers
 
 
@@ -113,10 +113,13 @@ def get(keys, missing=None, deferred=None, connection=None):
     connection = _require_connection(connection)
     dataset_id = _get_dataset_id_from_keys(keys)
 
+    transaction = Transaction.current()
+
     entity_pbs = connection.lookup(
         dataset_id=dataset_id,
         key_pbs=[k.to_protobuf() for k in keys],
         missing=missing, deferred=deferred,
+        transaction_id=transaction and transaction.id,
     )
 
     if missing is not None:
@@ -150,7 +153,7 @@ def put(entities, connection=None):
 
     connection = connection or _implicit_environ.CONNECTION
 
-    current = _BATCHES.top
+    current = Batch.current()
     in_batch = current is not None
     if not in_batch:
         keys = [entity.key for entity in entities]
@@ -177,7 +180,7 @@ def delete(keys, connection=None):
     connection = connection or _implicit_environ.CONNECTION
 
     # We allow partial keys to attempt a delete, the backend will fail.
-    current = _BATCHES.top
+    current = Batch.current()
     in_batch = current is not None
     if not in_batch:
         dataset_id = _get_dataset_id_from_keys(keys)
diff --git a/gcloud/datastore/batch.py b/gcloud/datastore/batch.py
index 5795e00809a9..7e0bb8f4cb62 100644
--- a/gcloud/datastore/batch.py
+++ b/gcloud/datastore/batch.py
@@ -130,6 +130,11 @@ def __init__(self, dataset_id=None, connection=None):
         self._mutation = datastore_pb.Mutation()
         self._auto_id_entities = []
 
+    @staticmethod
+    def current():
+        """Return the topmost batch / transaction, or None."""
+        return _BATCHES.top
+
     @property
     def dataset_id(self):
         """Getter for dataset ID in which the batch will run.
@@ -308,6 +313,11 @@ def _assign_entity_to_mutation(mutation_pb, entity, auto_id_entities):
     insert.key.CopyFrom(key_pb)
 
     for name, value in entity.items():
+
+        value_is_list = isinstance(value, list)
+        if value_is_list and len(value) == 0:
+            continue
+
         prop = insert.property.add()
         # Set the name of the property.
         prop.name = name
@@ -316,7 +326,7 @@ def _assign_entity_to_mutation(mutation_pb, entity, auto_id_entities):
         helpers._set_protobuf_value(prop.value, value)
 
         if name in entity.exclude_from_indexes:
-            if not isinstance(value, list):
+            if not value_is_list:
                 prop.value.indexed = False
 
             for sub_value in prop.value.list_value:
diff --git a/gcloud/datastore/connection.py b/gcloud/datastore/connection.py
index 15dbe980442b..402013a3332f 100644
--- a/gcloud/datastore/connection.py
+++ b/gcloud/datastore/connection.py
@@ -38,10 +38,6 @@ class Connection(connection.Connection):
                         '/datasets/{dataset_id}/{method}')
     """A template for the URL of a particular API call."""
 
-    def __init__(self, credentials=None):
-        super(Connection, self).__init__(credentials=credentials)
-        self._current_transaction = None
-
     def _request(self, dataset_id, method, data):
         """Make a request over the Http transport to the Cloud Datastore API.
 
@@ -128,37 +124,9 @@ def build_api_url(cls, dataset_id, method, base_url=None,
             api_version=(api_version or cls.API_VERSION),
             dataset_id=dataset_id, method=method)
 
-    def transaction(self, transaction=connection.Connection._EMPTY):
-        """Getter/setter for the connection's transaction object.
-
-        :type transaction: :class:`gcloud.datastore.transaction.Transaction`,
-                           (setting), or omitted (getting).
-        :param transaction: The new transaction (if passed).
-
-        :rtype: :class:`gcloud.datastore.transaction.Transaction`, (getting)
-                or :class:`gcloud.datastore.connection.Connection` (setting)
-        :returns: The current transaction (getting) or self (setting).
-        """
-        if transaction is self._EMPTY:
-            return self._current_transaction
-        else:
-            self._current_transaction = transaction
-            return self
-
-    def mutation(self):
-        """Getter for mutation usable with current connection.
-
-        :rtype: :class:`gcloud.datastore.datastore_v1_pb2.Mutation`.
-        :returns: The mutation instance associated with the current transaction
-                  (if one exists) or or a new mutation instance.
-        """
-        if self.transaction():
-            return self.transaction().mutation
-        else:
-            return datastore_pb.Mutation()
-
     def lookup(self, dataset_id, key_pbs,
-               missing=None, deferred=None, eventual=False):
+               missing=None, deferred=None,
+               eventual=False, transaction_id=None):
         """Lookup keys from a dataset in the Cloud Datastore.
 
         Maps the ``DatastoreService.Lookup`` protobuf RPC.
@@ -199,8 +167,12 @@ def lookup(self, dataset_id, key_pbs,
         :type eventual: boolean
         :param eventual: If False (the default), request ``STRONG`` read
                         consistency.  If True, request ``EVENTUAL`` read
-                        consistency.  If the connection has a current
-                        transaction, this value *must* be false.
+                        consistency.
+
+        :type transaction_id: string
+        :param transaction_id: If passed, make the request in the scope of
+                               the given transaction.  Incompatible with
+                               ``eventual==True``.
 
         :rtype: list of :class:`gcloud.datastore.datastore_v1_pb2.Entity`
                 (or a single Entity)
@@ -218,7 +190,7 @@ def lookup(self, dataset_id, key_pbs,
             raise ValueError('deferred must be None or an empty list')
 
         lookup_request = datastore_pb.LookupRequest()
-        self._set_read_options(lookup_request, eventual)
+        _set_read_options(lookup_request, eventual, transaction_id)
 
         single_key = isinstance(key_pbs, datastore_pb.Key)
 
@@ -244,7 +216,8 @@ def lookup(self, dataset_id, key_pbs,
 
         return results
 
-    def run_query(self, dataset_id, query_pb, namespace=None, eventual=False):
+    def run_query(self, dataset_id, query_pb, namespace=None,
+                  eventual=False, transaction_id=None):
         """Run a query on the Cloud Datastore.
 
         Maps the ``DatastoreService.RunQuery`` protobuf RPC.
@@ -293,11 +266,15 @@ def run_query(self, dataset_id, query_pb, namespace=None, eventual=False):
         :type eventual: boolean
         :param eventual: If False (the default), request ``STRONG`` read
                          consistency.  If True, request ``EVENTUAL`` read
-                         consistency.  If the connection has a current
-                         transaction, this value *must* be false.
+                         consistency.
+
+        :type transaction_id: string
+        :param transaction_id: If passed, make the request in the scope of
+                               the given transaction.  Incompatible with
+                               ``eventual==True``.
         """
         request = datastore_pb.RunQueryRequest()
-        self._set_read_options(request, eventual)
+        _set_read_options(request, eventual, transaction_id)
 
         if namespace:
             request.partition_id.namespace = namespace
@@ -328,10 +305,6 @@ def begin_transaction(self, dataset_id, serializable=False):
         :rtype: :class:`.datastore_v1_pb2.BeginTransactionResponse`
         :returns': the result protobuf for the begin transaction request.
         """
-        if self.transaction():
-            raise ValueError('Cannot start a transaction with another already '
-                             'in progress.')
-
         request = datastore_pb.BeginTransactionRequest()
 
         if serializable:
@@ -346,7 +319,7 @@ def begin_transaction(self, dataset_id, serializable=False):
 
         return response.transaction
 
-    def commit(self, dataset_id, mutation_pb):
+    def commit(self, dataset_id, mutation_pb, transaction_id=None):
         """Commit dataset mutations in context of current transation (if any).
 
         Maps the ``DatastoreService.Commit`` protobuf RPC.
@@ -357,14 +330,19 @@ def commit(self, dataset_id, mutation_pb):
         :type mutation_pb: :class:`gcloud.datastore.datastore_v1_pb2.Mutation`.
         :param mutation_pb: The protobuf for the mutations being saved.
 
+        :type transaction_id: string
+        :param transaction_id: The transaction ID returned from
+                               :meth:`begin_transaction`.  If not passed, the
+                               commit will be non-transactional.
+
         :rtype: :class:`gcloud.datastore.datastore_v1_pb2.MutationResult`.
         :returns': the result protobuf for the mutation.
         """
         request = datastore_pb.CommitRequest()
 
-        if self.transaction():
+        if transaction_id:
             request.mode = datastore_pb.CommitRequest.TRANSACTIONAL
-            request.transaction = self.transaction().id
+            request.transaction = transaction_id
         else:
             request.mode = datastore_pb.CommitRequest.NON_TRANSACTIONAL
 
@@ -373,7 +351,7 @@ def commit(self, dataset_id, mutation_pb):
                              datastore_pb.CommitResponse)
         return response.mutation_result
 
-    def rollback(self, dataset_id):
+    def rollback(self, dataset_id, transaction_id):
         """Rollback the connection's existing transaction.
 
         Maps the ``DatastoreService.Rollback`` protobuf RPC.
@@ -382,14 +360,12 @@ def rollback(self, dataset_id):
         :param dataset_id: The ID of the dataset to which the transaction
                            belongs.
 
-        :raises: :class:`ValueError` if the connection isn't currently in a
-                 transaction.
+        :type transaction_id: string
+        :param transaction_id: The transaction ID returned from
+                               :meth:`begin_transaction`.
         """
-        if not self.transaction() or not self.transaction().id:
-            raise ValueError('No transaction to rollback.')
-
         request = datastore_pb.RollbackRequest()
-        request.transaction = self.transaction().id
+        request.transaction = transaction_id
         # Nothing to do with this response, so just execute the method.
         self._rpc(dataset_id, 'rollback', request,
                   datastore_pb.RollbackResponse)
@@ -416,119 +392,6 @@ def allocate_ids(self, dataset_id, key_pbs):
                              datastore_pb.AllocateIdsResponse)
         return list(response.key)
 
-    def save_entity(self, dataset_id, key_pb, properties,
-                    exclude_from_indexes=(), mutation=None):
-        """Save an entity to the Cloud Datastore with the provided properties.
-
-        .. note::
-           Any existing properties for the entity identified by ``key_pb``
-           will be replaced by those passed in ``properties``;  properties
-           not passed in ``properties`` no longer be set for the entity.
-
-        .. note::
-           When saving an entity to the backend, a property value set as
-           an empty list cannot be saved and will be ignored.
-
-        :type dataset_id: string
-        :param dataset_id: The ID of the dataset in which to save the entity.
-
-        :type key_pb: :class:`gcloud.datastore.datastore_v1_pb2.Key`
-        :param key_pb: The complete or partial key for the entity.
-
-        :type properties: dict
-        :param properties: The properties to store on the entity.
-
-        :type exclude_from_indexes: sequence of string
-        :param exclude_from_indexes: Names of properties *not* to be indexed.
-
-        :type mutation: :class:`gcloud.datastore.datastore_v1_pb2.Mutation`
-                        or None.
-        :param mutation: If passed, the mutation protobuf into which the
-                         entity will be saved.  If None, use th result
-                         of calling ``self.mutation()``
-
-        :rtype: tuple
-        :returns: The pair (``assigned``, ``new_id``) where ``assigned`` is a
-                  boolean indicating if a new ID has been assigned and
-                  ``new_id`` is either ``None`` or an integer that has been
-                  assigned.
-        """
-        if mutation is not None:
-            in_batch = True
-        else:
-            in_batch = False
-            mutation = self.mutation()
-
-        key_pb = helpers._prepare_key_for_request(key_pb)
-
-        # If the Key is complete, we should upsert
-        # instead of using insert_auto_id.
-        path = key_pb.path_element[-1]
-        auto_id = not (path.HasField('id') or path.HasField('name'))
-
-        if auto_id:
-            insert = mutation.insert_auto_id.add()
-        else:
-            insert = mutation.upsert.add()
-
-        insert.key.CopyFrom(key_pb)
-
-        for name, value in properties.items():
-            helpers._set_protobuf_property(insert.property, name, value,
-                                           name not in exclude_from_indexes)
-
-        # If this is in a transaction, we should just return True. The
-        # transaction will handle assigning any keys as necessary.
-        if in_batch or self.transaction():
-            return False, None
-
-        result = self.commit(dataset_id, mutation)
-        # If this was an auto-assigned ID, return the new Key. We don't
-        # verify that this matches the original `key_pb` but trust the
-        # backend to uphold the values sent (e.g. dataset ID).
-        if auto_id:
-            inserted_key_pb = result.insert_auto_id_key[0]
-            # Assumes the backend has set `id` without checking HasField('id').
-            return True, inserted_key_pb.path_element[-1].id
-
-        return False, None
-
-    def delete_entities(self, dataset_id, key_pbs, mutation=None):
-        """Delete keys from a dataset in the Cloud Datastore.
-
-        This method deals only with
-        :class:`gcloud.datastore.datastore_v1_pb2.Key` protobufs and not
-        with any of the other abstractions.  For example, it's used
-        under the hood in :func:`gcloud.datastore.api.delete`.
-
-        :type dataset_id: string
-        :param dataset_id: The ID of the dataset from which to delete the keys.
-
-        :type key_pbs: list of :class:`gcloud.datastore.datastore_v1_pb2.Key`
-        :param key_pbs: The keys to delete from the datastore.
-
-        :type mutation: :class:`gcloud.datastore.datastore_v1_pb2.Mutation`
-                        or None.
-        :param mutation: If passed, the mutation protobuf into which the
-                         deletion will be saved.  If None, use th result
-                         of calling ``self.mutation()``
-
-        :rtype: boolean
-        :returns: ``True``
-        """
-        if mutation is not None:
-            in_batch = True
-        else:
-            in_batch = False
-            mutation = self.mutation()
-
-        helpers._add_keys_to_request(mutation.delete, key_pbs)
-
-        if not in_batch and not self.transaction():
-            self.commit(dataset_id, mutation)
-
-        return True
-
     def _lookup(self, lookup_request, dataset_id, stop_on_deferred):
         """Repeat lookup until all keys found (unless stop requested).
 
@@ -559,20 +422,20 @@ def _lookup(self, lookup_request, dataset_id, stop_on_deferred):
             _copy_deferred_keys(lookup_request, lookup_response)
         return results, missing, deferred
 
-    def _set_read_options(self, request, eventual):
-        """Validate rules for read options, and assign to the request.
 
-        Helper method for ``lookup()`` and ``run_query``.
-        """
-        transaction = self.transaction()
-        if eventual and transaction:
-            raise ValueError('eventual must be False when in a transaction')
-
-        opts = request.read_options
-        if eventual:
-            opts.read_consistency = datastore_pb.ReadOptions.EVENTUAL
-        elif transaction:
-            opts.transaction = transaction.id
+def _set_read_options(request, eventual, transaction_id):
+    """Validate rules for read options, and assign to the request.
+
+    Helper method for ``lookup()`` and ``run_query``.
+    """
+    if eventual and (transaction_id is not None):
+        raise ValueError('eventual must be False when in a transaction')
+
+    opts = request.read_options
+    if eventual:
+        opts.read_consistency = datastore_pb.ReadOptions.EVENTUAL
+    elif transaction_id:
+        opts.transaction = transaction_id
 
 
 def _copy_deferred_keys(lookup_request, lookup_response):
diff --git a/gcloud/datastore/helpers.py b/gcloud/datastore/helpers.py
index afee07891a1f..110afa546aed 100644
--- a/gcloud/datastore/helpers.py
+++ b/gcloud/datastore/helpers.py
@@ -218,44 +218,6 @@ def _get_value_from_property_pb(property_pb):
     return _get_value_from_value_pb(property_pb.value)
 
 
-def _set_protobuf_property(property_pb, name, value, indexed):
-    """Assign 'name', 'value', 'indexed' to the correct 'property_pb'.
-
-    Some value (empty list) cannot be directly assigned; this function handles
-    them correctly.
-
-    :type property_pb: :class:`gcloud.datastore.datastore_v1_pb2.Property`
-    :param property_pb: The value protobuf to which the value is being
-                        assigned.
-
-    :type name: string
-    :param name: The name to be assigned.
-
-    :type value: `datetime.datetime`, boolean, float, integer, string,
-                 :class:`gcloud.datastore.key.Key`,
-                 :class:`gcloud.datastore.entity.Entity`,
-    :param value: The value to be assigned.
-
-    :type indexed: boolean
-    :param indexed: The flag indicates the property should to be indexed or
-                    not.
-    """
-    if isinstance(value, list) and len(value) == 0:
-        return
-
-    prop = property_pb.add()
-    prop.name = name
-
-    _set_protobuf_value(prop.value, value)
-
-    if not indexed:
-        if not isinstance(value, list):
-            prop.value.indexed = False
-
-        for sub_value in prop.value.list_value:
-            sub_value.indexed = False
-
-
 def _set_protobuf_value(value_pb, val):
     """Assign 'val' to the correct subfield of 'value_pb'.
 
diff --git a/gcloud/datastore/query.py b/gcloud/datastore/query.py
index 58b2570a26cc..9b4ff7ee7a40 100644
--- a/gcloud/datastore/query.py
+++ b/gcloud/datastore/query.py
@@ -20,6 +20,7 @@
 from gcloud.datastore import datastore_v1_pb2 as datastore_pb
 from gcloud.datastore import helpers
 from gcloud.datastore.key import Key
+from gcloud.datastore.transaction import Transaction
 
 
 class Query(object):
@@ -377,10 +378,13 @@ def next_page(self):
 
         pb.offset = self._offset
 
+        transaction = Transaction.current()
+
         query_results = self._connection.run_query(
             query_pb=pb,
             dataset_id=self._query.dataset_id,
             namespace=self._query.namespace,
+            transaction_id=transaction and transaction.id,
             )
         # NOTE: `query_results` contains an extra value that we don't use,
         #       namely `skipped_results`.
diff --git a/gcloud/datastore/test_api.py b/gcloud/datastore/test_api.py
index c1ddc1527907..a8ad39b5855a 100644
--- a/gcloud/datastore/test_api.py
+++ b/gcloud/datastore/test_api.py
@@ -271,7 +271,7 @@ def test_hit_multiple_keys_different_dataset(self):
         with self.assertRaises(ValueError):
             self._callFUT([key1, key2], connection=object())
 
-    def test_implicit(self):
+    def test_implicit_wo_transaction(self):
         from gcloud.datastore import _implicit_environ
         from gcloud.datastore.key import Key
         from gcloud.datastore.test_connection import _Connection
@@ -297,6 +297,43 @@ def test_implicit(self):
         expected_called_with = {
             'dataset_id': DATASET_ID,
             'key_pbs': [key.to_protobuf()],
+            'transaction_id': None,
+        }
+        self.assertEqual(CUSTOM_CONNECTION._called_with, expected_called_with)
+
+        new_key = result.key
+        # Check the returned value is as expected.
+        self.assertFalse(new_key is key)
+        self.assertEqual(new_key.dataset_id, DATASET_ID)
+        self.assertEqual(new_key.path, PATH)
+        self.assertEqual(list(result), ['foo'])
+        self.assertEqual(result['foo'], 'Foo')
+
+    def test_w_transaction(self):
+        from gcloud.datastore.key import Key
+        from gcloud.datastore.test_connection import _Connection
+
+        DATASET_ID = 'DATASET'
+        KIND = 'Kind'
+        ID = 1234
+        PATH = [{'kind': KIND, 'id': ID}]
+        TRANSACTION = 'TRANSACTION'
+
+        # Make a found entity pb to be returned from mock backend.
+        entity_pb = self._make_entity_pb(DATASET_ID, KIND, ID,
+                                         'foo', 'Foo')
+
+        # Make a connection to return the entity pb.
+        CUSTOM_CONNECTION = _Connection(entity_pb)
+
+        key = Key(KIND, ID, dataset_id=DATASET_ID)
+        with _NoCommitTransaction(DATASET_ID, CUSTOM_CONNECTION, TRANSACTION):
+            result, = self._callFUT([key], connection=CUSTOM_CONNECTION)
+
+        expected_called_with = {
+            'dataset_id': DATASET_ID,
+            'key_pbs': [key.to_protobuf()],
+            'transaction_id': TRANSACTION,
         }
         self.assertEqual(CUSTOM_CONNECTION._called_with, expected_called_with)
 
@@ -340,10 +377,6 @@ def test_no_batch_w_partial_key(self):
         self.assertEqual(properties[0].value.string_value, u'bar')
 
     def test_existing_batch_w_completed_key(self):
-        from gcloud._testing import _Monkey
-        from gcloud.datastore import api
-        from gcloud.datastore.batch import _Batches
-        from gcloud.datastore.batch import Batch
         from gcloud.datastore.test_batch import _Connection
         from gcloud.datastore.test_batch import _Entity
         from gcloud.datastore.test_batch import _Key
@@ -354,12 +387,8 @@ def test_existing_batch_w_completed_key(self):
         entity = _Entity(foo=u'bar')
         key = entity.key = _Key(_DATASET)
 
-        # Set up mock Batch on stack so we can check it is used.
-        _BATCHES = _Batches()
-        CURR_BATCH = Batch(dataset_id=_DATASET, connection=connection)
-        _BATCHES.push(CURR_BATCH)
-
-        with _Monkey(api, _BATCHES=_BATCHES):
+        # Set up Batch on stack so we can check it is used.
+        with _NoCommitBatch(_DATASET, connection) as CURR_BATCH:
             result = self._callFUT([entity], connection=connection)
 
         self.assertEqual(result, None)
@@ -375,9 +404,6 @@ def test_existing_batch_w_completed_key(self):
     def test_implicit_connection(self):
         from gcloud._testing import _Monkey
         from gcloud.datastore import _implicit_environ
-        from gcloud.datastore import api
-        from gcloud.datastore.batch import _Batches
-        from gcloud.datastore.batch import Batch
         from gcloud.datastore.test_batch import _Connection
         from gcloud.datastore.test_batch import _Entity
         from gcloud.datastore.test_batch import _Key
@@ -388,13 +414,9 @@ def test_implicit_connection(self):
         entity = _Entity(foo=u'bar')
         key = entity.key = _Key(_DATASET)
 
-        # Set up mock Batch on stack so we can check it is used.
-        _BATCHES = _Batches()
-
         with _Monkey(_implicit_environ, CONNECTION=connection):
-            CURR_BATCH = Batch(dataset_id=_DATASET)
-            _BATCHES.push(CURR_BATCH)
-            with _Monkey(api, _BATCHES=_BATCHES):
+            # Set up Batch on stack so we can check it is used.
+            with _NoCommitBatch(_DATASET, connection) as CURR_BATCH:
                 result = self._callFUT([entity])
 
         self.assertEqual(result, None)
@@ -453,10 +475,6 @@ def test_no_batch(self):
         self.assertEqual(list(mutation.delete), [key.to_protobuf()])
 
     def test_existing_batch(self):
-        from gcloud._testing import _Monkey
-        from gcloud.datastore import api
-        from gcloud.datastore.batch import _Batches
-        from gcloud.datastore.batch import Batch
         from gcloud.datastore.test_batch import _Connection
         from gcloud.datastore.test_batch import _Key
 
@@ -465,12 +483,8 @@ def test_existing_batch(self):
         connection = _Connection()
         key = _Key(_DATASET)
 
-        # Set up mock Batch on stack so we can check it is used.
-        _BATCHES = _Batches()
-        CURR_BATCH = Batch(dataset_id=_DATASET, connection=connection)
-        _BATCHES.push(CURR_BATCH)
-
-        with _Monkey(api, _BATCHES=_BATCHES):
+        # Set up Batch on stack so we can check it is used.
+        with _NoCommitBatch(_DATASET, connection) as CURR_BATCH:
             result = self._callFUT([key], connection=connection)
 
         self.assertEqual(result, None)
@@ -484,9 +498,6 @@ def test_existing_batch(self):
     def test_implicit_connection(self):
         from gcloud._testing import _Monkey
         from gcloud.datastore import _implicit_environ
-        from gcloud.datastore import api
-        from gcloud.datastore.batch import _Batches
-        from gcloud.datastore.batch import Batch
         from gcloud.datastore.test_batch import _Connection
         from gcloud.datastore.test_batch import _Key
 
@@ -495,13 +506,9 @@ def test_implicit_connection(self):
         connection = _Connection()
         key = _Key(_DATASET)
 
-        # Set up mock Batch on stack so we can check it is used.
-        _BATCHES = _Batches()
-
         with _Monkey(_implicit_environ, CONNECTION=connection):
-            CURR_BATCH = Batch(dataset_id=_DATASET)
-            _BATCHES.push(CURR_BATCH)
-            with _Monkey(api, _BATCHES=_BATCHES):
+            # Set up Batch on stack so we can check it is used.
+            with _NoCommitBatch(_DATASET, connection) as CURR_BATCH:
                 result = self._callFUT([key])
 
         self.assertEqual(result, None)
@@ -583,3 +590,36 @@ def test_with_already_completed_key(self):
             COMPLETE_KEY = Key('KIND', 1234)
             self.assertRaises(ValueError, self._callFUT,
                               COMPLETE_KEY, 2)
+
+
+class _NoCommitBatch(object):
+
+    def __init__(self, dataset_id, connection):
+        from gcloud.datastore.batch import Batch
+        self._batch = Batch(dataset_id, connection)
+
+    def __enter__(self):
+        from gcloud.datastore.batch import _BATCHES
+        _BATCHES.push(self._batch)
+        return self._batch
+
+    def __exit__(self, *args):
+        from gcloud.datastore.batch import _BATCHES
+        _BATCHES.pop()
+
+
+class _NoCommitTransaction(object):
+
+    def __init__(self, dataset_id, connection, transaction_id):
+        from gcloud.datastore.transaction import Transaction
+        xact = self._transaction = Transaction(dataset_id, connection)
+        xact._id = transaction_id
+
+    def __enter__(self):
+        from gcloud.datastore.batch import _BATCHES
+        _BATCHES.push(self._transaction)
+        return self._transaction
+
+    def __exit__(self, *args):
+        from gcloud.datastore.batch import _BATCHES
+        _BATCHES.pop()
diff --git a/gcloud/datastore/test_batch.py b/gcloud/datastore/test_batch.py
index b1b911e8c4a1..24705f816cda 100644
--- a/gcloud/datastore/test_batch.py
+++ b/gcloud/datastore/test_batch.py
@@ -93,6 +93,24 @@ def test_ctor_implicit(self):
         self.assertTrue(isinstance(batch.mutation, Mutation))
         self.assertEqual(batch._auto_id_entities, [])
 
+    def test_current(self):
+        _DATASET = 'DATASET'
+        connection = _Connection()
+        batch1 = self._makeOne(_DATASET, connection)
+        batch2 = self._makeOne(_DATASET, connection)
+        self.assertTrue(batch1.current() is None)
+        self.assertTrue(batch2.current() is None)
+        with batch1:
+            self.assertTrue(batch1.current() is batch1)
+            self.assertTrue(batch2.current() is batch1)
+            with batch2:
+                self.assertTrue(batch1.current() is batch2)
+                self.assertTrue(batch2.current() is batch2)
+            self.assertTrue(batch1.current() is batch1)
+            self.assertTrue(batch2.current() is batch1)
+        self.assertTrue(batch1.current() is None)
+        self.assertTrue(batch2.current() is None)
+
     def test_add_auto_id_entity_w_partial_key(self):
         _DATASET = 'DATASET'
         connection = _Connection()
@@ -143,7 +161,12 @@ def test_put_entity_w_partial_key(self):
 
     def test_put_entity_w_completed_key(self):
         _DATASET = 'DATASET'
-        _PROPERTIES = {'foo': 'bar', 'baz': 'qux', 'spam': [1, 2, 3]}
+        _PROPERTIES = {
+            'foo': 'bar',
+            'baz': 'qux',
+            'spam': [1, 2, 3],
+            'frotz': [],  # will be ignored
+            }
         connection = _Connection()
         batch = self._makeOne(dataset_id=_DATASET, connection=connection)
         entity = _Entity(_PROPERTIES)
@@ -166,6 +189,7 @@ def test_put_entity_w_completed_key(self):
         self.assertFalse(props['spam'].list_value[0].indexed)
         self.assertFalse(props['spam'].list_value[1].indexed)
         self.assertFalse(props['spam'].list_value[2].indexed)
+        self.assertFalse('frotz' in props)
 
         deletes = list(batch.mutation.delete)
         self.assertEqual(len(deletes), 0)
diff --git a/gcloud/datastore/test_connection.py b/gcloud/datastore/test_connection.py
index 246e3863642d..194a15a387cb 100644
--- a/gcloud/datastore/test_connection.py
+++ b/gcloud/datastore/test_connection.py
@@ -179,40 +179,6 @@ def test_build_api_url_w_explicit_base_version(self):
         self.assertEqual(klass.build_api_url(DATASET_ID, METHOD, BASE, VER),
                          URI)
 
-    def test_transaction_getter_unset(self):
-        conn = self._makeOne()
-        self.assertTrue(conn.transaction() is None)
-
-    def test_transaction_setter(self):
-        xact = object()
-        conn = self._makeOne()
-        self.assertTrue(conn.transaction(xact) is conn)
-        self.assertTrue(conn.transaction() is xact)
-
-    def test_mutation_wo_transaction(self):
-        from gcloud._testing import _Monkey
-        from gcloud.datastore import datastore_v1_pb2 as datastore_pb
-
-        class Mutation(object):
-            pass
-        conn = self._makeOne()
-        with _Monkey(datastore_pb, Mutation=Mutation):
-            found = conn.mutation()
-        self.assertTrue(isinstance(found, Mutation))
-
-    def test_mutation_w_transaction(self):
-
-        class Mutation(object):
-            pass
-
-        class Xact(object):
-            mutation = Mutation()
-
-        conn = self._makeOne()
-        conn.transaction(Xact())
-        found = conn.mutation()
-        self.assertTrue(isinstance(found, Mutation))
-
     def test_lookup_single_key_empty_response(self):
         from gcloud.datastore import datastore_v1_pb2 as datastore_pb
 
@@ -273,9 +239,8 @@ def test_lookup_single_key_empty_response_w_eventual_and_transaction(self):
         TRANSACTION = 'TRANSACTION'
         key_pb = self._make_key_pb(DATASET_ID)
         conn = self._makeOne()
-        conn.transaction(Transaction(TRANSACTION))
-        self.assertRaises(
-            ValueError, conn.lookup, DATASET_ID, key_pb, eventual=True)
+        self.assertRaises(ValueError, conn.lookup, DATASET_ID, key_pb,
+                          eventual=True, transaction_id=TRANSACTION)
 
     def test_lookup_single_key_empty_response_w_transaction(self):
         from gcloud.datastore import datastore_v1_pb2 as datastore_pb
@@ -285,7 +250,6 @@ def test_lookup_single_key_empty_response_w_transaction(self):
         key_pb = self._make_key_pb(DATASET_ID)
         rsp_pb = datastore_pb.LookupResponse()
         conn = self._makeOne()
-        conn.transaction(Transaction(TRANSACTION))
         URI = '/'.join([
             conn.API_BASE_URL,
             'datastore',
@@ -295,7 +259,8 @@ def test_lookup_single_key_empty_response_w_transaction(self):
             'lookup',
         ])
         http = conn._http = Http({'status': '200'}, rsp_pb.SerializeToString())
-        self.assertEqual(conn.lookup(DATASET_ID, key_pb), None)
+        found = conn.lookup(DATASET_ID, key_pb, transaction_id=TRANSACTION)
+        self.assertEqual(found, None)
         cw = http._called_with
         self._verifyProtobufCall(cw, URI, conn)
         rq_class = datastore_pb.LookupRequest
@@ -564,7 +529,6 @@ def test_run_query_wo_eventual_w_transaction(self):
         rsp_pb.batch.more_results = no_more
         rsp_pb.batch.entity_result_type = datastore_pb.EntityResult.FULL
         conn = self._makeOne()
-        conn.transaction(Transaction(TRANSACTION))
         URI = '/'.join([
             conn.API_BASE_URL,
             'datastore',
@@ -574,7 +538,8 @@ def test_run_query_wo_eventual_w_transaction(self):
             'runQuery',
         ])
         http = conn._http = Http({'status': '200'}, rsp_pb.SerializeToString())
-        pbs, end, more, skipped = conn.run_query(DATASET_ID, q_pb)
+        pbs, end, more, skipped = conn.run_query(
+            DATASET_ID, q_pb, transaction_id=TRANSACTION)
         self.assertEqual(pbs, [])
         self.assertEqual(end, CURSOR)
         self.assertTrue(more)
@@ -604,9 +569,8 @@ def test_run_query_w_eventual_and_transaction(self):
         rsp_pb.batch.more_results = no_more
         rsp_pb.batch.entity_result_type = datastore_pb.EntityResult.FULL
         conn = self._makeOne()
-        conn.transaction(Transaction(TRANSACTION))
-        self.assertRaises(
-            ValueError, conn.run_query, DATASET_ID, q_pb, eventual=True)
+        self.assertRaises(ValueError, conn.run_query, DATASET_ID, q_pb,
+                          eventual=True, transaction_id=TRANSACTION)
 
     def test_run_query_wo_namespace_empty_result(self):
         from gcloud.datastore import datastore_v1_pb2 as datastore_pb
@@ -674,12 +638,6 @@ def test_run_query_w_namespace_nonempty_result(self):
         self.assertEqual(request.partition_id.namespace, 'NS')
         self.assertEqual(request.query, q_pb)
 
-    def test_begin_transaction_w_existing_transaction(self):
-        DATASET_ID = 'DATASET'
-        conn = self._makeOne()
-        conn.transaction(object())
-        self.assertRaises(ValueError, conn.begin_transaction, DATASET_ID)
-
     def test_begin_transaction_default_serialize(self):
         from gcloud.datastore import datastore_v1_pb2 as datastore_pb
 
@@ -767,9 +725,6 @@ def test_commit_wo_transaction(self):
     def test_commit_w_transaction(self):
         from gcloud.datastore import datastore_v1_pb2 as datastore_pb
 
-        class Xact(object):
-            id = 'xact'
-
         DATASET_ID = 'DATASET'
         key_pb = self._make_key_pb(DATASET_ID)
         rsp_pb = datastore_pb.CommitResponse()
@@ -780,7 +735,6 @@ class Xact(object):
         prop.name = 'foo'
         prop.value.string_value = u'Foo'
         conn = self._makeOne()
-        conn.transaction(Xact())
         URI = '/'.join([
             conn.API_BASE_URL,
             'datastore',
@@ -790,7 +744,7 @@ class Xact(object):
             'commit',
         ])
         http = conn._http = Http({'status': '200'}, rsp_pb.SerializeToString())
-        result = conn.commit(DATASET_ID, mutation)
+        result = conn.commit(DATASET_ID, mutation, 'xact')
         self.assertEqual(result.index_updates, 0)
         self.assertEqual(list(result.insert_auto_id_key), [])
         cw = http._called_with
@@ -802,34 +756,13 @@ class Xact(object):
         self.assertEqual(request.mutation, mutation)
         self.assertEqual(request.mode, rq_class.TRANSACTIONAL)
 
-    def test_rollback_wo_existing_transaction(self):
-        DATASET_ID = 'DATASET'
-        conn = self._makeOne()
-        self.assertRaises(ValueError,
-                          conn.rollback, DATASET_ID)
-
-    def test_rollback_w_existing_transaction_no_id(self):
-
-        class Xact(object):
-            id = None
-
-        DATASET_ID = 'DATASET'
-        conn = self._makeOne()
-        conn.transaction(Xact())
-        self.assertRaises(ValueError,
-                          conn.rollback, DATASET_ID)
-
     def test_rollback_ok(self):
         from gcloud.datastore import datastore_v1_pb2 as datastore_pb
         DATASET_ID = 'DATASET'
         TRANSACTION = 'xact'
 
-        class Xact(object):
-            id = TRANSACTION
-
         rsp_pb = datastore_pb.RollbackResponse()
         conn = self._makeOne()
-        conn.transaction(Xact())
         URI = '/'.join([
             conn.API_BASE_URL,
             'datastore',
@@ -839,7 +772,7 @@ class Xact(object):
             'rollback',
         ])
         http = conn._http = Http({'status': '200'}, rsp_pb.SerializeToString())
-        self.assertEqual(conn.rollback(DATASET_ID), None)
+        self.assertEqual(conn.rollback(DATASET_ID, TRANSACTION), None)
         cw = http._called_with
         self._verifyProtobufCall(cw, URI, conn)
         rq_class = datastore_pb.RollbackRequest
@@ -906,299 +839,6 @@ def test_allocate_ids_non_empty(self):
         for key_before, key_after in zip(before_key_pbs, request.key):
             _compare_key_pb_after_request(self, key_before, key_after)
 
-    def test_save_entity_w_empty_list_value(self):
-        from gcloud.datastore import datastore_v1_pb2 as datastore_pb
-
-        DATASET_ID = 'DATASET'
-        key_pb = self._make_key_pb(DATASET_ID)
-        rsp_pb = datastore_pb.CommitResponse()
-        conn = self._makeOne()
-        URI = '/'.join([
-            conn.API_BASE_URL,
-            'datastore',
-            conn.API_VERSION,
-            'datasets',
-            DATASET_ID,
-            'commit',
-        ])
-        http = conn._http = Http({'status': '200'}, rsp_pb.SerializeToString())
-        result = conn.save_entity(DATASET_ID, key_pb,
-                                  {'foo': u'Foo', 'bar': []})
-        self.assertEqual(result, (False, None))
-        cw = http._called_with
-        self._verifyProtobufCall(cw, URI, conn)
-        rq_class = datastore_pb.CommitRequest
-        request = rq_class()
-        request.ParseFromString(cw['body'])
-        self.assertEqual(request.transaction, '')
-        mutation = request.mutation
-        self.assertEqual(len(mutation.insert_auto_id), 0)
-        upserts = list(mutation.upsert)
-        self.assertEqual(len(upserts), 1)
-        upsert = upserts[0]
-        _compare_key_pb_after_request(self, key_pb, upsert.key)
-        props = list(upsert.property)
-        self.assertEqual(len(props), 1)
-        self.assertEqual(props[0].name, 'foo')
-        self.assertEqual(props[0].value.string_value, u'Foo')
-        self.assertEqual(props[0].value.indexed, True)
-        self.assertEqual(len(mutation.delete), 0)
-        self.assertEqual(request.mode, rq_class.NON_TRANSACTIONAL)
-
-    def test_save_entity_wo_transaction_w_upsert(self):
-        from gcloud.datastore import datastore_v1_pb2 as datastore_pb
-
-        DATASET_ID = 'DATASET'
-        key_pb = self._make_key_pb(DATASET_ID)
-        rsp_pb = datastore_pb.CommitResponse()
-        conn = self._makeOne()
-        URI = '/'.join([
-            conn.API_BASE_URL,
-            'datastore',
-            conn.API_VERSION,
-            'datasets',
-            DATASET_ID,
-            'commit',
-        ])
-        http = conn._http = Http({'status': '200'}, rsp_pb.SerializeToString())
-        result = conn.save_entity(DATASET_ID, key_pb, {'foo': u'Foo'})
-        self.assertEqual(result, (False, None))
-        cw = http._called_with
-        self._verifyProtobufCall(cw, URI, conn)
-        rq_class = datastore_pb.CommitRequest
-        request = rq_class()
-        request.ParseFromString(cw['body'])
-        self.assertEqual(request.transaction, '')
-        mutation = request.mutation
-        self.assertEqual(len(mutation.insert_auto_id), 0)
-        upserts = list(mutation.upsert)
-        self.assertEqual(len(upserts), 1)
-        upsert = upserts[0]
-        _compare_key_pb_after_request(self, key_pb, upsert.key)
-        props = list(upsert.property)
-        self.assertEqual(len(props), 1)
-        self.assertEqual(props[0].name, 'foo')
-        self.assertEqual(props[0].value.string_value, u'Foo')
-        self.assertEqual(props[0].value.indexed, True)
-        self.assertEqual(len(mutation.delete), 0)
-        self.assertEqual(request.mode, rq_class.NON_TRANSACTIONAL)
-
-    def test_save_entity_w_exclude_from_indexes(self):
-        from gcloud.datastore import datastore_v1_pb2 as datastore_pb
-        import operator
-
-        DATASET_ID = 'DATASET'
-        key_pb = self._make_key_pb(DATASET_ID)
-        rsp_pb = datastore_pb.CommitResponse()
-        conn = self._makeOne()
-        URI = '/'.join([
-            conn.API_BASE_URL,
-            'datastore',
-            conn.API_VERSION,
-            'datasets',
-            DATASET_ID,
-            'commit',
-        ])
-        http = conn._http = Http({'status': '200'}, rsp_pb.SerializeToString())
-        result = conn.save_entity(DATASET_ID, key_pb,
-                                  {'foo': u'Foo', 'bar': [u'bar1', u'bar2']},
-                                  exclude_from_indexes=['foo', 'bar'])
-        self.assertEqual(result, (False, None))
-        cw = http._called_with
-        self._verifyProtobufCall(cw, URI, conn)
-        rq_class = datastore_pb.CommitRequest
-        request = rq_class()
-        request.ParseFromString(cw['body'])
-        self.assertEqual(request.transaction, '')
-        mutation = request.mutation
-        self.assertEqual(len(mutation.insert_auto_id), 0)
-        upserts = list(mutation.upsert)
-        self.assertEqual(len(upserts), 1)
-        upsert = upserts[0]
-        _compare_key_pb_after_request(self, key_pb, upsert.key)
-        props = sorted(upsert.property,
-                       key=operator.attrgetter('name'),
-                       reverse=True)
-        self.assertEqual(len(props), 2)
-        self.assertEqual(props[0].name, 'foo')
-        self.assertEqual(props[0].value.string_value, u'Foo')
-        self.assertEqual(props[0].value.indexed, False)
-        self.assertEqual(props[1].name, 'bar')
-        self.assertEqual(props[1].value.list_value[0].string_value, 'bar1')
-        self.assertEqual(props[1].value.list_value[1].string_value, 'bar2')
-        self.assertEqual(props[1].value.HasField('indexed'), False)
-        self.assertEqual(props[1].value.list_value[0].indexed, False)
-        self.assertEqual(props[1].value.list_value[1].indexed, False)
-        self.assertEqual(len(mutation.delete), 0)
-        self.assertEqual(request.mode, rq_class.NON_TRANSACTIONAL)
-
-    def test_save_entity_wo_transaction_w_auto_id(self):
-        from gcloud.datastore import datastore_v1_pb2 as datastore_pb
-
-        DATASET_ID = 'DATASET'
-        key_pb = self._make_key_pb(DATASET_ID, id=None)
-        updated_key_pb = self._make_key_pb(DATASET_ID)
-        rsp_pb = datastore_pb.CommitResponse()
-        mr_pb = rsp_pb.mutation_result
-        mr_pb.index_updates = 0
-        iaik_pb = mr_pb.insert_auto_id_key.add()
-        iaik_pb.CopyFrom(updated_key_pb)
-        conn = self._makeOne()
-        URI = '/'.join([
-            conn.API_BASE_URL,
-            'datastore',
-            conn.API_VERSION,
-            'datasets',
-            DATASET_ID,
-            'commit',
-        ])
-        http = conn._http = Http({'status': '200'}, rsp_pb.SerializeToString())
-        result = conn.save_entity(DATASET_ID, key_pb, {'foo': u'Foo'})
-        self.assertEqual(result, (True, 1234))
-        cw = http._called_with
-        self._verifyProtobufCall(cw, URI, conn)
-        rq_class = datastore_pb.CommitRequest
-        request = rq_class()
-        request.ParseFromString(cw['body'])
-        self.assertEqual(request.transaction, '')
-        mutation = request.mutation
-        inserts = list(mutation.insert_auto_id)
-        insert = inserts[0]
-        _compare_key_pb_after_request(self, key_pb, insert.key)
-        props = list(insert.property)
-        self.assertEqual(len(props), 1)
-        self.assertEqual(props[0].name, 'foo')
-        self.assertEqual(props[0].value.string_value, u'Foo')
-        self.assertEqual(len(inserts), 1)
-        upserts = list(mutation.upsert)
-        self.assertEqual(len(upserts), 0)
-        self.assertEqual(len(mutation.delete), 0)
-        self.assertEqual(request.mode, rq_class.NON_TRANSACTIONAL)
-
-    def test_save_entity_w_transaction(self):
-        from gcloud.datastore import datastore_v1_pb2 as datastore_pb
-
-        class Xact(object):
-            mutation = datastore_pb.Mutation()
-
-        DATASET_ID = 'DATASET'
-        key_pb = self._make_key_pb(DATASET_ID)
-        rsp_pb = datastore_pb.CommitResponse()
-        conn = self._makeOne()
-        conn.transaction(Xact())
-        http = conn._http = Http({'status': '200'}, rsp_pb.SerializeToString())
-        result = conn.save_entity(DATASET_ID, key_pb, {'foo': u'Foo'})
-        self.assertEqual(result, (False, None))
-        self.assertEqual(http._called_with, None)
-        mutation = conn.mutation()
-        self.assertEqual(len(mutation.upsert), 1)
-
-    def test_save_entity_w_transaction_nested_entity(self):
-        from gcloud.datastore import datastore_v1_pb2 as datastore_pb
-        from gcloud.datastore.entity import Entity
-
-        class Xact(object):
-            mutation = datastore_pb.Mutation()
-
-        DATASET_ID = 'DATASET'
-        nested = Entity()
-        nested['bar'] = u'Bar'
-        key_pb = self._make_key_pb(DATASET_ID)
-        rsp_pb = datastore_pb.CommitResponse()
-        conn = self._makeOne()
-        conn.transaction(Xact())
-        http = conn._http = Http({'status': '200'}, rsp_pb.SerializeToString())
-        result = conn.save_entity(DATASET_ID, key_pb, {'foo': nested})
-        self.assertEqual(result, (False, None))
-        self.assertEqual(http._called_with, None)
-        mutation = conn.mutation()
-        self.assertEqual(len(mutation.upsert), 1)
-
-    def test_save_entity_w_mutation_passed(self):
-        from gcloud.datastore import datastore_v1_pb2 as datastore_pb
-        from gcloud.datastore.entity import Entity
-        DATASET_ID = 'DATASET'
-        nested = Entity()
-        nested['bar'] = u'Bar'
-        key_pb = self._make_key_pb(DATASET_ID)
-        rsp_pb = datastore_pb.CommitResponse()
-        conn = self._makeOne()
-        http = conn._http = Http({'status': '200'}, rsp_pb.SerializeToString())
-        mutation = datastore_pb.Mutation()
-        result = conn.save_entity(DATASET_ID, key_pb, {'foo': nested},
-                                  mutation=mutation)
-        self.assertEqual(result, (False, None))
-        self.assertEqual(http._called_with, None)
-        conn_mutation = conn.mutation()
-        self.assertEqual(len(mutation.upsert), 1)
-        self.assertEqual(len(conn_mutation.upsert), 0)
-
-    def test_delete_entities_wo_transaction(self):
-        from gcloud.datastore import datastore_v1_pb2 as datastore_pb
-
-        DATASET_ID = 'DATASET'
-        key_pb = self._make_key_pb(DATASET_ID)
-        rsp_pb = datastore_pb.CommitResponse()
-        conn = self._makeOne()
-        URI = '/'.join([
-            conn.API_BASE_URL,
-            'datastore',
-            conn.API_VERSION,
-            'datasets',
-            DATASET_ID,
-            'commit',
-        ])
-        http = conn._http = Http({'status': '200'}, rsp_pb.SerializeToString())
-        result = conn.delete_entities(DATASET_ID, [key_pb])
-        self.assertEqual(result, True)
-        cw = http._called_with
-        self._verifyProtobufCall(cw, URI, conn)
-        rq_class = datastore_pb.CommitRequest
-        request = rq_class()
-        request.ParseFromString(cw['body'])
-        self.assertEqual(request.transaction, '')
-        mutation = request.mutation
-        self.assertEqual(len(mutation.insert_auto_id), 0)
-        self.assertEqual(len(mutation.upsert), 0)
-        deletes = list(mutation.delete)
-        self.assertEqual(len(deletes), 1)
-        delete = deletes[0]
-        _compare_key_pb_after_request(self, key_pb, delete)
-        self.assertEqual(request.mode, rq_class.NON_TRANSACTIONAL)
-
-    def test_delete_entities_w_transaction(self):
-        from gcloud.datastore import datastore_v1_pb2 as datastore_pb
-
-        class Xact(object):
-            mutation = datastore_pb.Mutation()
-
-        DATASET_ID = 'DATASET'
-        key_pb = self._make_key_pb(DATASET_ID)
-        rsp_pb = datastore_pb.CommitResponse()
-        conn = self._makeOne()
-        conn.transaction(Xact())
-        http = conn._http = Http({'status': '200'}, rsp_pb.SerializeToString())
-        result = conn.delete_entities(DATASET_ID, [key_pb])
-        self.assertEqual(result, True)
-        self.assertEqual(http._called_with, None)
-        mutation = conn.mutation()
-        self.assertEqual(len(mutation.delete), 1)
-
-    def test_delete_entities_w_mutation_passed(self):
-        from gcloud.datastore import datastore_v1_pb2 as datastore_pb
-        DATASET_ID = 'DATASET'
-        key_pb = self._make_key_pb(DATASET_ID)
-        rsp_pb = datastore_pb.CommitResponse()
-        conn = self._makeOne()
-        http = conn._http = Http({'status': '200'}, rsp_pb.SerializeToString())
-        mutation = datastore_pb.Mutation()
-        result = conn.delete_entities(DATASET_ID, [key_pb], mutation=mutation)
-        self.assertEqual(result, True)
-        self.assertEqual(http._called_with, None)
-        conn_mutation = conn.mutation()
-        self.assertEqual(len(mutation.delete), 1)
-        self.assertEqual(len(conn_mutation.delete), 0)
-
 
 class Http(object):
 
@@ -1225,16 +865,6 @@ def request(self, **kw):
         return result
 
 
-class Transaction(object):
-
-    def __init__(self, id):
-        self._id = id
-
-    @property
-    def id(self):
-        return self._id
-
-
 def _compare_key_pb_after_request(test, key_before, key_after):
     test.assertFalse(key_after.partition_id.HasField('dataset_id'))
     test.assertEqual(key_before.partition_id.namespace,
diff --git a/gcloud/datastore/test_query.py b/gcloud/datastore/test_query.py
index d843fa7ee014..971348bd2fa7 100644
--- a/gcloud/datastore/test_query.py
+++ b/gcloud/datastore/test_query.py
@@ -405,6 +405,7 @@ def test_next_page_no_cursors_no_more(self):
             'dataset_id': self._DATASET,
             'query_pb': qpb,
             'namespace': self._NAMESPACE,
+            'transaction_id': None,
         }
         self.assertEqual(connection._called_with, [EXPECTED])
 
@@ -431,6 +432,7 @@ def test_next_page_no_cursors_no_more_w_offset_and_limit(self):
             'dataset_id': self._DATASET,
             'query_pb': qpb,
             'namespace': self._NAMESPACE,
+            'transaction_id': None,
         }
         self.assertEqual(connection._called_with, [EXPECTED])
 
@@ -463,6 +465,7 @@ def test_next_page_w_cursors_w_more(self):
             'dataset_id': self._DATASET,
             'query_pb': qpb,
             'namespace': self._NAMESPACE,
+            'transaction_id': None,
         }
         self.assertEqual(connection._called_with, [EXPECTED])
 
@@ -494,6 +497,7 @@ def test___iter___no_more(self):
             'dataset_id': self._DATASET,
             'query_pb': qpb,
             'namespace': self._NAMESPACE,
+            'transaction_id': None,
         }
         self.assertEqual(connection._called_with, [EXPECTED])
 
@@ -522,11 +526,13 @@ def test___iter___w_more(self):
             'dataset_id': self._DATASET,
             'query_pb': qpb1,
             'namespace': self._NAMESPACE,
+            'transaction_id': None,
         }
         EXPECTED2 = {
             'dataset_id': self._DATASET,
             'query_pb': qpb2,
             'namespace': self._NAMESPACE,
+            'transaction_id': None,
         }
         self.assertEqual(len(connection._called_with), 2)
         self.assertEqual(connection._called_with[0], EXPECTED1)
diff --git a/gcloud/datastore/test_transaction.py b/gcloud/datastore/test_transaction.py
index 75093367d8c3..ead71196eb39 100644
--- a/gcloud/datastore/test_transaction.py
+++ b/gcloud/datastore/test_transaction.py
@@ -65,6 +65,31 @@ def test_ctor_with_env(self):
         self.assertEqual(xact.dataset_id, DATASET_ID)
         self.assertEqual(xact.connection, CONNECTION)
 
+    def test_current(self):
+        from gcloud.datastore.test_api import _NoCommitBatch
+        _DATASET = 'DATASET'
+        connection = _Connection()
+        xact1 = self._makeOne(_DATASET, connection)
+        xact2 = self._makeOne(_DATASET, connection)
+        self.assertTrue(xact1.current() is None)
+        self.assertTrue(xact2.current() is None)
+        with xact1:
+            self.assertTrue(xact1.current() is xact1)
+            self.assertTrue(xact2.current() is xact1)
+            with _NoCommitBatch(_DATASET, _Connection):
+                self.assertTrue(xact1.current() is None)
+                self.assertTrue(xact2.current() is None)
+            with xact2:
+                self.assertTrue(xact1.current() is xact2)
+                self.assertTrue(xact2.current() is xact2)
+                with _NoCommitBatch(_DATASET, _Connection):
+                    self.assertTrue(xact1.current() is None)
+                    self.assertTrue(xact2.current() is None)
+            self.assertTrue(xact1.current() is xact1)
+            self.assertTrue(xact2.current() is xact1)
+        self.assertTrue(xact1.current() is None)
+        self.assertTrue(xact2.current() is None)
+
     def test_begin(self):
         _DATASET = 'DATASET'
         connection = _Connection(234)
@@ -72,7 +97,6 @@ def test_begin(self):
         xact.begin()
         self.assertEqual(xact.id, 234)
         self.assertEqual(connection._begun, _DATASET)
-        self.assertTrue(connection._xact is xact)
 
     def test_rollback(self):
         _DATASET = 'DATASET'
@@ -82,7 +106,6 @@ def test_rollback(self):
         xact.rollback()
         self.assertEqual(xact.id, None)
         self.assertEqual(connection._rolled_back, _DATASET)
-        self.assertEqual(connection._xact, None)
 
     def test_commit_no_auto_ids(self):
         _DATASET = 'DATASET'
@@ -92,7 +115,6 @@ def test_commit_no_auto_ids(self):
         xact.begin()
         xact.commit()
         self.assertEqual(connection._committed, (_DATASET, mutation))
-        self.assertTrue(connection._xact is None)
         self.assertEqual(xact.id, None)
 
     def test_commit_w_auto_ids(self):
@@ -109,22 +131,9 @@ def test_commit_w_auto_ids(self):
         xact.begin()
         xact.commit()
         self.assertEqual(connection._committed, (_DATASET, mutation))
-        self.assertTrue(connection._xact is None)
         self.assertEqual(xact.id, None)
         self.assertEqual(entity.key.path, [{'kind': _KIND, 'id': _ID}])
 
-    def test_commit_w_already(self):
-        _DATASET = 'DATASET'
-        connection = _Connection(234)
-        xact = self._makeOne(dataset_id=_DATASET, connection=connection)
-        xact._mutation = object()
-        xact.begin()
-        connection.transaction(())  # Simulate previous commit via false-ish.
-        xact.commit()
-        self.assertEqual(connection._committed, None)
-        self.assertTrue(connection._xact is None)
-        self.assertEqual(xact.id, None)
-
     def test_context_manager_no_raise(self):
         _DATASET = 'DATASET'
         connection = _Connection(234)
@@ -133,9 +142,7 @@ def test_context_manager_no_raise(self):
         with xact:
             self.assertEqual(xact.id, 234)
             self.assertEqual(connection._begun, _DATASET)
-            self.assertTrue(connection._xact is xact)
         self.assertEqual(connection._committed, (_DATASET, mutation))
-        self.assertTrue(connection._xact is None)
         self.assertEqual(xact.id, None)
 
     def test_context_manager_w_raise(self):
@@ -149,14 +156,11 @@ class Foo(Exception):
             with xact:
                 self.assertEqual(xact.id, 234)
                 self.assertEqual(connection._begun, _DATASET)
-                self.assertTrue(connection._xact is xact)
                 raise Foo()
         except Foo:
             self.assertEqual(xact.id, None)
             self.assertEqual(connection._rolled_back, _DATASET)
-            self.assertEqual(connection._xact, None)
         self.assertEqual(connection._committed, None)
-        self.assertTrue(connection._xact is None)
         self.assertEqual(xact.id, None)
 
 
@@ -173,17 +177,12 @@ def _make_key(kind, id, dataset_id):
 
 class _Connection(object):
     _marker = object()
-    _begun = _rolled_back = _committed = _xact = None
+    _begun = _rolled_back = _committed = None
 
     def __init__(self, xact_id=123):
         self._xact_id = xact_id
         self._commit_result = _CommitResult()
 
-    def transaction(self, xact=_marker):
-        if xact is self._marker:
-            return self._xact
-        self._xact = xact
-
     def begin_transaction(self, dataset_id):
         self._begun = dataset_id
         return self._xact_id
diff --git a/gcloud/datastore/transaction.py b/gcloud/datastore/transaction.py
index 4b0b93f8d0ba..8378a4f89bb7 100644
--- a/gcloud/datastore/transaction.py
+++ b/gcloud/datastore/transaction.py
@@ -73,15 +73,6 @@ class Transaction(Batch):
          ...     assert entity.key.is_partial  # There is no ID on this key.
          >>> assert not entity.key.is_partial  # There *is* an ID.
 
-    .. warning:: If you're using the automatically generated ID
-       functionality, it's important that you only use
-       :meth:`gcloud.datastore.entity.Entity.save` rather than using
-       :meth:`gcloud.datastore.connection.Connection.save_entity`
-       directly.
-
-       If you mix the two, the results will have extra IDs generated and
-       it could jumble things up.
-
     If you don't want to use the context manager you can initialize a
     transaction manually::
 
@@ -119,6 +110,19 @@ def id(self):
         """
         return self._id
 
+    @staticmethod
+    def current():
+        """Return the topmost transaction.
+
+        .. note:: if the topmost element on the stack is not a transaction,
+                  returns None.
+
+        :rtype: :class:`gcloud.datastore.transaction.Transaction` or None
+        """
+        top = Batch.current()
+        if isinstance(top, Transaction):
+            return top
+
     def begin(self):
         """Begins a transaction.
 
@@ -127,7 +131,6 @@ def begin(self):
         to use a context manager.
         """
         self._id = self.connection.begin_transaction(self._dataset_id)
-        self.connection.transaction(self)
 
     def rollback(self):
         """Rolls back the current transaction.
@@ -138,7 +141,6 @@ def rollback(self):
         - Sets the current transaction's ID to None.
         """
         self.connection.rollback(self._dataset_id)
-        self.connection.transaction(None)
         self._id = None
 
     def commit(self):
@@ -150,17 +152,9 @@ def commit(self):
 
         This method has necessary side-effects:
 
-        - Sets the current connection's transaction reference to None.
         - Sets the current transaction's ID to None.
-        - Updates paths for any keys that needed an automatically generated ID.
         """
-        # It's possible that they called commit() already, in which case
-        # we shouldn't do any committing of our own.
-        if self.connection.transaction():
-            super(Transaction, self).commit()
-
-        # Tell the connection that the transaction is over.
-        self.connection.transaction(None)
+        super(Transaction, self).commit()
 
         # Clear our own ID in case this gets accidentally reused.
         self._id = None