Skip to content

Commit a1194f5

Browse files
dhermesdhermes
committed
Using protobuf CommitRequest in datastore Connection.commit.
This is towards #1288 in preparation for the upgrade to `v1beta3`. In particular, a single `Mutation` protobuf instance in `v1beta3` is not sufficient to contain all changes to be committed, so we use the container that is up one level in the hierarchy.
1 parent f220a36 commit a1194f5

File tree

6 files changed

+54
-45
lines changed

6 files changed

+54
-45
lines changed

‎gcloud/datastore/batch.py‎

Lines changed: 17 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -60,11 +60,12 @@ class Batch(object):
6060
:type client: :class:`gcloud.datastore.client.Client`
6161
:param client: The client used to connect to datastore.
6262
"""
63+
6364
_id = None # "protected" attribute, always None for non-transactions
6465

6566
def __init__(self, client):
6667
self._client = client
67-
self._mutation = _datastore_pb2.Mutation()
68+
self._commit_request = _datastore_pb2.CommitRequest()
6869
self._partial_key_entities = []
6970

7071
def current(self):
@@ -114,6 +115,9 @@ def _add_complete_key_entity_pb(self):
114115
:returns: The newly created entity protobuf that will be
115116
updated and sent with a commit.
116117
"""
118+
# We use ``upsert`` for entities with completed keys, rather than
119+
# ``insert`` or ``update``, in order not to create race conditions
120+
# based on prior existence / removal of the entity.
117121
return self.mutations.upsert.add()
118122

119123
def _add_delete_key_pb(self):
@@ -129,17 +133,16 @@ def _add_delete_key_pb(self):
129133
def mutations(self):
130134
"""Getter for the changes accumulated by this batch.
131135
132-
Every batch is committed with a single Mutation
133-
representing the 'work' to be done as part of the batch.
134-
Inside a batch, calling ``batch.put()`` with an entity, or
135-
``batch.delete`` with a key, builds up the mutation.
136-
This getter returns the Mutation protobuf that
137-
has been built-up so far.
136+
Every batch is committed with a single commit request containing all
137+
the work to be done as mutations. Inside a batch, calling :meth:`put`
138+
with an entity, or :meth:`delete` with a key, builds up the request by
139+
adding a new mutation. This getter returns the protobuf that has been
140+
built-up so far.
138141
139142
:rtype: :class:`gcloud.datastore._generated.datastore_pb2.Mutation`
140143
:returns: The Mutation protobuf to be sent in the commit request.
141144
"""
142-
return self._mutation
145+
return self._commit_request.mutation
143146

144147
def put(self, entity):
145148
"""Remember an entity's state to be saved during ``commit``.
@@ -155,9 +158,9 @@ def put(self, entity):
155158
Python3) map to 'string_value' in the datastore; values which are
156159
"bytes" ('str' in Python2, 'bytes' in Python3) map to 'blob_value'.
157160
158-
When an entity has a partial key, calling :meth:`commit`` sends it as
159-
an ``insert_auto_id`` mutation and the key is completed. On return, the
160-
key for the ``entity`` passed in as updated to match the key ID
161+
When an entity has a partial key, calling :meth:`commit` sends it as
162+
an ``insert_auto_id`` mutation and the key is completed. On return,
163+
the key for the ``entity`` passed in is updated to match the key ID
161164
assigned by the server.
162165
163166
:type entity: :class:`gcloud.datastore.entity.Entity`
@@ -212,11 +215,10 @@ def commit(self):
212215
context manager.
213216
"""
214217
_, updated_keys = self.connection.commit(
215-
self.dataset_id, self.mutations, self._id)
218+
self.dataset_id, self._commit_request, self._id)
216219
# If the back-end returns without error, we are guaranteed that
217-
# the response's 'insert_auto_id_key' will match (length and order)
218-
# the request's 'insert_auto_id` entities, which are derived from
219-
# our '_partial_key_entities' (no partial success).
220+
# :meth:`Connection.commit` will return keys that match (length and
221+
# order) directly ``_partial_key_entities``.
220222
for new_key_pb, entity in zip(updated_keys,
221223
self._partial_key_entities):
222224
new_id = new_key_pb.path_element[-1].id

‎gcloud/datastore/connection.py‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -290,16 +290,16 @@ def begin_transaction(self, dataset_id):
290290
_datastore_pb2.BeginTransactionResponse)
291291
return response.transaction
292292

293-
def commit(self, dataset_id, mutation_pb, transaction_id):
294-
"""Commit dataset mutations in context of current transation (if any).
293+
def commit(self, dataset_id, commit_request, transaction_id):
294+
"""Commit mutations in context of current transation (if any).
295295
296296
Maps the ``DatastoreService.Commit`` protobuf RPC.
297297
298298
:type dataset_id: string
299299
:param dataset_id: The ID dataset to which the transaction applies.
300300
301-
:type mutation_pb: :class:`._generated.datastore_pb2.Mutation`
302-
:param mutation_pb: The protobuf for the mutations being saved.
301+
:type commit_request: :class:`._generated.datastore_pb2.CommitRequest`
302+
:param commit_request: The protobuf with the mutations being committed.
303303
304304
:type transaction_id: string or None
305305
:param transaction_id: The transaction ID returned from
@@ -312,14 +312,14 @@ def commit(self, dataset_id, mutation_pb, transaction_id):
312312
that was completed in the commit.
313313
"""
314314
request = _datastore_pb2.CommitRequest()
315+
request.CopyFrom(commit_request)
315316

316317
if transaction_id:
317318
request.mode = _datastore_pb2.CommitRequest.TRANSACTIONAL
318319
request.transaction = transaction_id
319320
else:
320321
request.mode = _datastore_pb2.CommitRequest.NON_TRANSACTIONAL
321322

322-
request.mutation.CopyFrom(mutation_pb)
323323
response = self._rpc(dataset_id, 'commit', request,
324324
_datastore_pb2.CommitResponse)
325325
return _parse_commit_response(response)

‎gcloud/datastore/test_batch.py‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -203,7 +203,7 @@ def test_commit(self):
203203
batch.commit()
204204

205205
self.assertEqual(connection._committed,
206-
[(_DATASET, batch.mutations, None)])
206+
[(_DATASET, batch._commit_request, None)])
207207

208208
def test_commit_w_partial_key_entities(self):
209209
_DATASET = 'DATASET'
@@ -219,7 +219,7 @@ def test_commit_w_partial_key_entities(self):
219219
batch.commit()
220220

221221
self.assertEqual(connection._committed,
222-
[(_DATASET, batch.mutations, None)])
222+
[(_DATASET, batch._commit_request, None)])
223223
self.assertFalse(entity.key.is_partial)
224224
self.assertEqual(entity.key._id, _NEW_ID)
225225

@@ -242,7 +242,7 @@ def test_as_context_mgr_wo_error(self):
242242
mutated_entity = _mutated_pb(self, batch.mutations, 'upsert')
243243
self.assertEqual(mutated_entity.key, key._key)
244244
self.assertEqual(connection._committed,
245-
[(_DATASET, batch.mutations, None)])
245+
[(_DATASET, batch._commit_request, None)])
246246

247247
def test_as_context_mgr_nested(self):
248248
_DATASET = 'DATASET'
@@ -274,8 +274,8 @@ def test_as_context_mgr_nested(self):
274274
self.assertEqual(mutated_entity2.key, key2._key)
275275

276276
self.assertEqual(connection._committed,
277-
[(_DATASET, batch2.mutations, None),
278-
(_DATASET, batch1.mutations, None)])
277+
[(_DATASET, batch2._commit_request, None),
278+
(_DATASET, batch1._commit_request, None)])
279279

280280
def test_as_context_mgr_w_error(self):
281281
_DATASET = 'DATASET'
@@ -323,8 +323,8 @@ def __init__(self, *new_keys):
323323
self._committed = []
324324
self._index_updates = 0
325325

326-
def commit(self, dataset_id, mutation, transaction_id):
327-
self._committed.append((dataset_id, mutation, transaction_id))
326+
def commit(self, dataset_id, commit_request, transaction_id):
327+
self._committed.append((dataset_id, commit_request, transaction_id))
328328
return self._index_updates, self._completed_keys
329329

330330

‎gcloud/datastore/test_client.py‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -624,9 +624,10 @@ def test_put_multi_no_batch_w_partial_key(self):
624624
self.assertTrue(result is None)
625625

626626
self.assertEqual(len(client.connection._commit_cw), 1)
627-
dataset_id, mutation, transaction_id = client.connection._commit_cw[0]
627+
(dataset_id,
628+
commit_req, transaction_id) = client.connection._commit_cw[0]
628629
self.assertEqual(dataset_id, self.DATASET_ID)
629-
inserts = list(mutation.insert_auto_id)
630+
inserts = list(commit_req.mutation.insert_auto_id)
630631
self.assertEqual(len(inserts), 1)
631632
self.assertEqual(inserts[0].key, key.to_protobuf())
632633
properties = list(inserts[0].property)
@@ -689,9 +690,10 @@ def test_delete_multi_no_batch(self):
689690
result = client.delete_multi([key])
690691
self.assertEqual(result, None)
691692
self.assertEqual(len(client.connection._commit_cw), 1)
692-
dataset_id, mutation, transaction_id = client.connection._commit_cw[0]
693+
(dataset_id,
694+
commit_req, transaction_id) = client.connection._commit_cw[0]
693695
self.assertEqual(dataset_id, self.DATASET_ID)
694-
self.assertEqual(list(mutation.delete), [key.to_protobuf()])
696+
self.assertEqual(list(commit_req.mutation.delete), [key.to_protobuf()])
695697
self.assertTrue(transaction_id is None)
696698

697699
def test_delete_multi_w_existing_batch(self):
@@ -1004,8 +1006,8 @@ def lookup(self, dataset_id, key_pbs, eventual=False, transaction_id=None):
10041006
results, missing, deferred = triple
10051007
return results, missing, deferred
10061008

1007-
def commit(self, dataset_id, mutation, transaction_id):
1008-
self._commit_cw.append((dataset_id, mutation, transaction_id))
1009+
def commit(self, dataset_id, commit_request, transaction_id):
1010+
self._commit_cw.append((dataset_id, commit_request, transaction_id))
10091011
response, self._commit = self._commit[0], self._commit[1:]
10101012
return self._index_updates, response
10111013

‎gcloud/datastore/test_connection.py‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -674,7 +674,8 @@ def test_commit_wo_transaction(self):
674674
DATASET_ID = 'DATASET'
675675
key_pb = self._make_key_pb(DATASET_ID)
676676
rsp_pb = datastore_pb2.CommitResponse()
677-
mutation = datastore_pb2.Mutation()
677+
req_pb = datastore_pb2.CommitRequest()
678+
mutation = req_pb.mutation
678679
insert = mutation.upsert.add()
679680
insert.key.CopyFrom(key_pb)
680681
prop = insert.property.add()
@@ -700,7 +701,7 @@ def mock_parse(response):
700701
return expected_result
701702

702703
with _Monkey(MUT, _parse_commit_response=mock_parse):
703-
result = conn.commit(DATASET_ID, mutation, None)
704+
result = conn.commit(DATASET_ID, req_pb, None)
704705

705706
self.assertTrue(result is expected_result)
706707
cw = http._called_with
@@ -721,7 +722,8 @@ def test_commit_w_transaction(self):
721722
DATASET_ID = 'DATASET'
722723
key_pb = self._make_key_pb(DATASET_ID)
723724
rsp_pb = datastore_pb2.CommitResponse()
724-
mutation = datastore_pb2.Mutation()
725+
req_pb = datastore_pb2.CommitRequest()
726+
mutation = req_pb.mutation
725727
insert = mutation.upsert.add()
726728
insert.key.CopyFrom(key_pb)
727729
prop = insert.property.add()
@@ -747,7 +749,7 @@ def mock_parse(response):
747749
return expected_result
748750

749751
with _Monkey(MUT, _parse_commit_response=mock_parse):
750-
result = conn.commit(DATASET_ID, mutation, b'xact')
752+
result = conn.commit(DATASET_ID, req_pb, b'xact')
751753

752754
self.assertTrue(result is expected_result)
753755
cw = http._called_with

‎gcloud/datastore/test_transaction.py‎

Lines changed: 11 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -102,10 +102,11 @@ def test_commit_no_partial_keys(self):
102102
connection = _Connection(234)
103103
client = _Client(_DATASET, connection)
104104
xact = self._makeOne(client)
105-
xact._mutation = mutation = object()
105+
xact._commit_request = commit_request = object()
106106
xact.begin()
107107
xact.commit()
108-
self.assertEqual(connection._committed, (_DATASET, mutation, 234))
108+
self.assertEqual(connection._committed,
109+
(_DATASET, commit_request, 234))
109110
self.assertEqual(xact.id, None)
110111

111112
def test_commit_w_partial_keys(self):
@@ -118,10 +119,11 @@ def test_commit_w_partial_keys(self):
118119
xact = self._makeOne(client)
119120
entity = _Entity()
120121
xact.put(entity)
121-
xact._mutation = mutation = object()
122+
xact._commit_request = commit_request = object()
122123
xact.begin()
123124
xact.commit()
124-
self.assertEqual(connection._committed, (_DATASET, mutation, 234))
125+
self.assertEqual(connection._committed,
126+
(_DATASET, commit_request, 234))
125127
self.assertEqual(xact.id, None)
126128
self.assertEqual(entity.key.path, [{'kind': _KIND, 'id': _ID}])
127129

@@ -130,11 +132,12 @@ def test_context_manager_no_raise(self):
130132
connection = _Connection(234)
131133
client = _Client(_DATASET, connection)
132134
xact = self._makeOne(client)
133-
xact._mutation = mutation = object()
135+
xact._commit_request = commit_request = object()
134136
with xact:
135137
self.assertEqual(xact.id, 234)
136138
self.assertEqual(connection._begun, _DATASET)
137-
self.assertEqual(connection._committed, (_DATASET, mutation, 234))
139+
self.assertEqual(connection._committed,
140+
(_DATASET, commit_request, 234))
138141
self.assertEqual(xact.id, None)
139142

140143
def test_context_manager_w_raise(self):
@@ -186,8 +189,8 @@ def begin_transaction(self, dataset_id):
186189
def rollback(self, dataset_id, transaction_id):
187190
self._rolled_back = dataset_id, transaction_id
188191

189-
def commit(self, dataset_id, mutation, transaction_id):
190-
self._committed = (dataset_id, mutation, transaction_id)
192+
def commit(self, dataset_id, commit_request, transaction_id):
193+
self._committed = (dataset_id, commit_request, transaction_id)
191194
return self._index_updates, self._completed_keys
192195

193196

0 commit comments

Comments
 (0)