Add UpdateByQueryResponse.success()

Author: sethmlarson

elasticsearch_dsl/response/__init__.py

@@ -113,3 +113,6 @@ def __init__(self, search, response, doc_class=None):
         super(AttrDict, self).__setattr__("_search", search)
         super(AttrDict, self).__setattr__("_doc_class", doc_class)
         super(UpdateByQueryResponse, self).__init__(response)
+
+    def success(self):
+        return not self.timed_out and not self.failures

tests/test_integration/test_update_by_query.py

@@ -35,6 +35,7 @@ def test_update_by_query_no_script(write_client, setup_ubq_tests):
     assert response.updated == 52
     assert response.deleted == 0
     assert response.took > 0
+    assert response.success()
 
 
 def test_update_by_query_with_script(write_client, setup_ubq_tests):
@@ -69,3 +70,4 @@ def test_delete_by_query_with_script(write_client, setup_ubq_tests):
 
     assert response.total == 1
     assert response.deleted == 1
+    assert response.success()

tests/test_update_by_query.py

@@ -18,6 +18,7 @@
 from copy import deepcopy
 
 from elasticsearch_dsl import Q, UpdateByQuery
+from elasticsearch_dsl.response import UpdateByQueryResponse
 
 
 def test_ubq_starts_with_no_query():
@@ -159,3 +160,14 @@ def test_overwrite_script():
     } == ubq.to_dict()
     ubq = ubq.script(source="ctx._source.likes++")
     assert {"script": {"source": "ctx._source.likes++"}} == ubq.to_dict()
+
+
+def test_update_by_query_response_success():
+    ubqr = UpdateByQueryResponse({}, {"timed_out": False, "failures": []})
+    assert ubqr.success()
+
+    ubqr = UpdateByQueryResponse({}, {"timed_out": True, "failures": []})
+    assert not ubqr.success()
+
+    ubqr = UpdateByQueryResponse({}, {"timed_out": False, "failures": [{}]})
+    assert not ubqr.success()