Set application/x-ndjson content type on bulk requests.

Resolves #609

elasticsearch/client/__init__.py

@@ -1127,7 +1127,8 @@ class Elasticsearch(object):
         if body in SKIP_IN_PATH:
             raise ValueError("Empty value passed for a required argument 'body'.")
         return self.transport.perform_request('POST', _make_path(index,
-            doc_type, '_bulk'), params=params, body=self._bulk_body(body))
+            doc_type, '_bulk'), params=params, body=self._bulk_body(body),
+            headers={'Content-Type': 'application/x-ndjson'})
 
     @query_params('max_concurrent_searches', 'pre_filter_shard_size',
         'search_type', 'typed_keys')

elasticsearch/connection/base.py

@@ -123,5 +123,3 @@ class Connection(object):
             logger.warning('Undecodable raw error response from server: %s', err)
 
         raise HTTP_EXCEPTIONS.get(status_code, TransportError)(status_code, error_message, additional_info)
-
-

elasticsearch/connection/http_requests.py

@@ -61,13 +61,13 @@ class RequestsHttpConnection(Connection):
             warnings.warn(
                 'Connecting to %s using SSL with verify_certs=False is insecure.' % self.base_url)
 
-    def perform_request(self, method, url, params=None, body=None, timeout=None, ignore=()):
+    def perform_request(self, method, url, params=None, body=None, timeout=None, ignore=(), headers=None):
         url = self.base_url + url
         if params:
             url = '%s?%s' % (url, urlencode(params or {}))
 
         start = time.time()
-        request = requests.Request(method=method, url=url, data=body)
+        request = requests.Request(method=method, headers=headers, url=url, data=body)
         prepared_request = self.session.prepare_request(request)
         settings = self.session.merge_environment_settings(prepared_request.url, {}, None, None, None)
         send_kwargs = {'timeout': timeout or self.timeout}

elasticsearch/connection/http_urllib3.py

@@ -91,7 +91,7 @@ class Urllib3HttpConnection(Connection):
 
         self.pool = pool_class(host, port=port, timeout=self.timeout, maxsize=maxsize, **kw)
 
-    def perform_request(self, method, url, params=None, body=None, timeout=None, ignore=()):
+    def perform_request(self, method, url, params=None, body=None, timeout=None, ignore=(), headers=None):
         url = self.url_prefix + url
         if params:
             url = '%s?%s' % (url, urlencode(params))
@@ -111,6 +111,8 @@ class Urllib3HttpConnection(Connection):
             if not isinstance(method, str):
                 method = method.encode('utf-8')
 
+            request_headers = dict(self.headers)
+            request_headers.update(headers or {})
             response = self.pool.urlopen(method, url, body, retries=False, headers=self.headers, **kw)
             duration = time.time() - start
             raw_data = response.data.decode('utf-8')

elasticsearch/transport.py

@@ -255,7 +255,7 @@ class Transport(object):
         if self.sniff_on_connection_fail:
             self.sniff_hosts()
 
-    def perform_request(self, method, url, params=None, body=None):
+    def perform_request(self, method, url, headers=None, params=None, body=None):
         """
         Perform the actual request. Retrieve a connection from the connection
         pool, pass all the information to it's perform_request method and
@@ -269,6 +269,8 @@ class Transport(object):
 
         :arg method: HTTP method to use
         :arg url: absolute url (without host) to target
+        :arg headers: dictionary of headers, will be handed over to the
+            underlying :class:`~elasticsearch.Connection` class for serialization
         :arg params: dictionary of query parameters, will be handed over to the
             underlying :class:`~elasticsearch.Connection` class for serialization
         :arg body: body of the request, will be serializes using serializer and
@@ -292,7 +294,7 @@ class Transport(object):
 
         if body is not None:
             try:
-                body = body.encode('utf-8', 'surrogatepass')            
+                body = body.encode('utf-8', 'surrogatepass')
             except (UnicodeDecodeError, AttributeError):
                 # bytes/str - no need to re-encode
                 pass
@@ -309,7 +311,7 @@ class Transport(object):
             connection = self.get_connection()
 
             try:
-                status, headers, data = connection.perform_request(method, url, params, body, ignore=ignore, timeout=timeout)
+                status, headers, data = connection.perform_request(method, url, params, body, headers=headers, ignore=ignore, timeout=timeout)
 
             except TransportError as e:
                 if method == 'HEAD' and e.status_code == 404:

test_elasticsearch/test_transport.py

@@ -74,7 +74,7 @@ class TestTransport(TestCase):
         t.perform_request('GET', '/', params={'request_timeout': 42})
         self.assertEquals(1, len(t.get_connection().calls))
         self.assertEquals(('GET', '/', {}, None), t.get_connection().calls[0][0])
-        self.assertEquals({'timeout': 42, 'ignore': ()}, t.get_connection().calls[0][1])
+        self.assertEquals({'timeout': 42, 'ignore': (), 'headers': None}, t.get_connection().calls[0][1])
 
     def test_send_get_body_as_source(self):
         t = Transport([{}], send_get_body_as='source', connection_class=DummyConnection)