Move _bulk_body to client.utils

elasticsearch/client/__init__.py

@@ -14,7 +14,7 @@ from .remote import RemoteClient
 from .snapshot import SnapshotClient
 from .tasks import TasksClient
 from .xpack import XPackClient
-from .utils import query_params, _make_path, SKIP_IN_PATH
+from .utils import query_params, _make_path, SKIP_IN_PATH, _bulk_body
 
 # xpack APIs
 from .ccr import CcrClient
@@ -268,17 +268,6 @@ class Elasticsearch(object):
             # probably operating on custom transport and connection_pool, ignore
             return super(Elasticsearch, self).__repr__()
 
-    def _bulk_body(self, body):
-        # if not passed in a string, serialize items and join by newline
-        if not isinstance(body, string_types):
-            body = "\n".join(map(self.transport.serializer.dumps, body))
-
-        # bulk body must end with a newline
-        if not body.endswith("\n"):
-            body += "\n"
-
-        return body
-
     # AUTO-GENERATED-API-DEFINITIONS #
     @query_params()
     def ping(self, params=None):
@@ -455,7 +444,7 @@ class Elasticsearch(object):
         if body in SKIP_IN_PATH:
             raise ValueError("Empty value passed for a required argument 'body'.")
 
-        body = self._bulk_body(body)
+        body = _bulk_body(self.transport.serializer, body)
         return self.transport.perform_request(
             "POST", _make_path(index, doc_type, "_bulk"), params=params, body=body
         )
@@ -1141,7 +1130,7 @@ class Elasticsearch(object):
         if body in SKIP_IN_PATH:
             raise ValueError("Empty value passed for a required argument 'body'.")
 
-        body = self._bulk_body(body)
+        body = _bulk_body(self.transport.serializer, body)
         return self.transport.perform_request(
             "GET", _make_path(index, doc_type, "_msearch"), params=params, body=body
         )
@@ -1173,7 +1162,7 @@ class Elasticsearch(object):
         if body in SKIP_IN_PATH:
             raise ValueError("Empty value passed for a required argument 'body'.")
 
-        body = self._bulk_body(body)
+        body = _bulk_body(self.transport.serializer, body)
         return self.transport.perform_request(
             "GET",
             _make_path(index, doc_type, "_msearch", "template"),

elasticsearch/client/ml.py

@@ -1,4 +1,4 @@
-from .utils import NamespacedClient, query_params, _make_path, SKIP_IN_PATH
+from .utils import NamespacedClient, query_params, _make_path, SKIP_IN_PATH, _bulk_body
 
 
 class MlClient(NamespacedClient):
@@ -229,7 +229,7 @@ class MlClient(NamespacedClient):
         if body in SKIP_IN_PATH:
             raise ValueError("Empty value passed for a required argument 'body'.")
 
-        body = self._bulk_body(body)
+        body = _bulk_body(self.transport.serializer, body)
         return self.transport.perform_request(
             "POST", "/_ml/find_file_structure", params=params, body=body
         )
@@ -685,7 +685,7 @@ class MlClient(NamespacedClient):
             if param in SKIP_IN_PATH:
                 raise ValueError("Empty value passed for a required argument.")
 
-        body = self._bulk_body(body)
+        body = _bulk_body(self.transport.serializer, body)
         return self.transport.perform_request(
             "POST",
             _make_path("_ml", "anomaly_detectors", job_id, "_data"),

elasticsearch/client/utils.py

@@ -88,6 +88,18 @@ def query_params(*es_query_params):
     return _wrapper
 
 
+def _bulk_body(serializer, body):
+    # if not passed in a string, serialize items and join by newline
+    if not isinstance(body, string_types):
+        body = "\n".join(map(serializer.dumps, body))
+
+    # bulk body must end with a newline
+    if not body.endswith("\n"):
+        body += "\n"
+
+    return body
+
+
 class NamespacedClient(object):
     def __init__(self, client):
         self.client = client