Make BulkIndexError and ScanError serializable

elasticsearch/helpers/errors.py

@@ -23,10 +23,17 @@ def __init__(self, message: Any, errors: List[Dict[str, Any]]):
  super().__init__(message)
  self.errors: List[Dict[str, Any]] = errors
 
+ def __reduce__(self):
+ return (self.__class__, (self.args[0], self.errors))
+
 
 class ScanError(Exception):
  scroll_id: str
 
  def __init__(self, scroll_id: str, *args: Any, **kwargs: Any) -> None:
  super().__init__(*args, **kwargs)
  self.scroll_id = scroll_id
+
+ def __reduce__(self):
+ return (self.__class__, (self.scroll_id,) + self.args)
+

test_elasticsearch/test_server/test_helpers.py

@@ -16,6 +16,7 @@
 # under the License.
 
 import json
+import pickle
 from datetime import datetime, timedelta
 from unittest.mock import call, patch
 
@@ -24,7 +25,7 @@
 from elastic_transport import ApiResponseMeta, ObjectApiResponse
 
 from elasticsearch import ApiError, helpers
-from elasticsearch.helpers import ScanError
+from elasticsearch.helpers import BulkIndexError, ScanError
 
 
 class FailingBulkClient:
@@ -993,3 +994,23 @@ def test_reindex_index_datastream_op_type_index(sync_client):
  query={"query": {"bool": {"filter": {"term": {"type": "answers"}}}}},
  op_type="_index",
  )
+
+def test_serialize_bulk_index_error():
+ msg = "message"
+ errors = {"error": 1}
+ error = BulkIndexError(msg, errors)
+ pickled = pickle.dumps(error)
+ actual = pickle.loads(pickled)
+ assert actual.__class__ == BulkIndexError
+ assert actual.errors == error.errors
+ assert actual.args == error.args
+
+def test_serialize_scan_error():
+ scroll_id = "message"
+ args = ("a", "b")
+ error = ScanError(scroll_id, *args)
+ pickled = pickle.dumps(error)
+ actual = pickle.loads(pickled)
+ assert actual.__class__ == ScanError
+ assert actual.scroll_id == error.scroll_id
+ assert actual.args == error.args