Merge pull request #618 from jmcarp/issue-609-x-ndjson

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

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')
@@ -1159,7 +1160,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('GET', _make_path(index,
-            doc_type, '_msearch'), params=params, body=self._bulk_body(body))
+            doc_type, '_msearch'), params=params, body=self._bulk_body(body),
+            headers={'content-type': 'application/x-ndjson'})
 
     @query_params('field_statistics', 'fields', 'offsets', 'parent', 'payloads',
         'positions', 'preference', 'realtime', 'routing', 'term_statistics',
@@ -1363,7 +1365,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('GET', _make_path(index, doc_type,
-            '_msearch', 'template'), params=params, body=self._bulk_body(body))
+            '_msearch', 'template'), params=params, body=self._bulk_body(body),
+            headers={'content-type': 'application/x-ndjson'})
 
     @query_params('allow_no_indices', 'expand_wildcards', 'fields',
         'ignore_unavailable')
@@ -1387,3 +1390,4 @@ class Elasticsearch(object):
         """
         return self.transport.perform_request('GET', _make_path(index,
             '_field_caps'), params=params, body=body)
+

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,9 @@ class Urllib3HttpConnection(Connection):
             if not isinstance(method, str):
                 method = method.encode('utf-8')
 
+            if headers:
+                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
         :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
@@ -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_connection.py

@@ -104,6 +104,13 @@ class TestRequestsConnection(TestCase):
         self.assertEquals('GET', request.method)
         self.assertEquals(None, request.body)
 
+    def test_merge_headers(self):
+        con = self._get_mock_connection(connection_params={'headers': {'h1': 'v1', 'h2': 'v2'}})
+        req = self._get_request(con, 'GET', '/', headers={'h2': 'v2p', 'h3': 'v3'})
+        self.assertEquals(req.headers['h1'], 'v1')
+        self.assertEquals(req.headers['h2'], 'v2p')
+        self.assertEquals(req.headers['h3'], 'v3')
+
     def test_http_auth(self):
         con = RequestsHttpConnection(http_auth='username:secret')
         self.assertEquals(('username', 'secret'), con.session.auth)

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)