Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Feat: Threaded MutationsBatcher #722

Merged
merged 24 commits into from
May 10, 2023
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
24 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Update MutationsBatchError
  • Loading branch information
Mariatta committed Apr 5, 2023
commit fd06c2ff0e8352cbc4c5c5d46c35e47cb0e0bdae
17 changes: 9 additions & 8 deletions google/cloud/bigtable/batcher.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,13 +37,9 @@
class MutationsBatchError(Exception):
"""Error in the batch request"""
Mariatta marked this conversation as resolved.
Show resolved Hide resolved

def __init__(self, status_codes):
self.errors = [
from_grpc_status(status_code.code, status_code.message)
for status_code in status_codes
if status_code.code != 0
]
self.message = "Errors in batch mutations."
def __init__(self, message, exc):
self.exc = exc
self.message = message
super().__init__(self.message)


Expand Down Expand Up @@ -312,7 +308,12 @@ def flush_rows(self, rows_to_flush=None):
responses.append(result)

if has_error:
raise MutationsBatchError(status_codes=responses)
exc = [
from_grpc_status(status_code.code, status_code.message)
for status_code in responses
if status_code.code != 0
]
raise MutationsBatchError(message="Errors in batch mutations.", exc=exc)

return responses

Expand Down
12 changes: 6 additions & 6 deletions tests/unit/test_batcher.py
Original file line number Diff line number Diff line change
Expand Up @@ -243,10 +243,10 @@ def test_mutations_batcher_response_with_error_codes():
with pytest.raises(MutationsBatchError) as exc:
mutation_batcher.flush()
assert exc.value.message == "Errors in batch mutations."
assert len(exc.value.errors) == 2
assert len(exc.value.exc) == 2

assert exc.value.errors[0].message == mocked_response[0].message
assert exc.value.errors[1].message == mocked_response[1].message
assert exc.value.exc[0].message == mocked_response[0].message
assert exc.value.exc[1].message == mocked_response[1].message


def test_mutations_batcher_flush_async_raises_exception():
Expand All @@ -266,10 +266,10 @@ def test_mutations_batcher_flush_async_raises_exception():
with pytest.raises(MutationsBatchError) as exc:
mutation_batcher.flush_async()
assert exc.value.message == "Errors in batch mutations."
assert len(exc.value.errors) == 2
assert len(exc.value.exc) == 2

assert exc.value.errors[0].message == mocked_response[0].message
assert exc.value.errors[1].message == mocked_response[1].message
assert exc.value.exc[0].message == mocked_response[0].message
assert exc.value.exc[1].message == mocked_response[1].message


class _Instance(object):
Expand Down