Repository: elastic/elasticsearch-py

Issue:
Set `application/x-ndjson` content type on bulk requests
As of elastic 5.x, requests to `/_bulk` should set `Content-Type` to `application/x-ndjson`. If not, elastic logs a warning. It looks like this library defaults to `application/json`. To fix, I'm thinking we should accept an optional dict of headers at `https://github.com/elastic/elasticsearch-py/blob/master/elasticsearch/connection/http_urllib3.py#L94`. If that sounds reasonable, I'd be happy to submit a patch.

cc @cnelson @wjwoodson

Discussion:
davidt99: @jmcarp Any plan to fix the review comments? Most of the driver requests (if not all) doesn't state content type, resulting warning logs in elastic 5.5.
The header support you wrote will be allow adding content type to all the request, thus fixing the warning.
HonzaKral: @davidt99 All requests are anotated with content-type accepted by elasticsearch since d862c8a (released in `5.2`). You can also set headers globally by specifying `headers={'custom': 'header'}` when instantiating `Elasticsearch`.
davidt99: @HonzaKral Oh, I was using 5.1 and looked for similar issue. Thanks.
jmcarp: I cleaned this up based on your suggestions @HonzaKral; let me know if this needs more revisions. I'm not sure why the tests are failing, but it looks like they're failing in the same way as `master`.
jmcarp: I rebased on master and dropped the deprecated methods, and reverted whitespace changes. This is ready for another look when you have time @HonzaKral.
fxdgear: This LGTM. Can you please rebase off master and push again? I had to change the version of ES that travis uses. 

Patch:
diff --git a/elasticsearch/client/__init__.py b/elasticsearch/client/__init__.py
index 59858549..834f1d86 100644
--- a/elasticsearch/client/__init__.py
+++ b/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)
+
diff --git a/elasticsearch/connection/http_requests.py b/elasticsearch/connection/http_requests.py
index 59dd381c..b98e7772 100644
--- a/elasticsearch/connection/http_requests.py
+++ b/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}
diff --git a/elasticsearch/connection/http_urllib3.py b/elasticsearch/connection/http_urllib3.py
index 7b4e6c79..62957ed2 100644
--- a/elasticsearch/connection/http_urllib3.py
+++ b/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')
diff --git a/elasticsearch/transport.py b/elasticsearch/transport.py
index dc8cd891..f876a945 100644
--- a/elasticsearch/transport.py
+++ b/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 patch:
diff --git a/test_elasticsearch/test_connection.py b/test_elasticsearch/test_connection.py
index b2d84996..e4e0ae63 100644
--- a/test_elasticsearch/test_connection.py
+++ b/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)
diff --git a/test_elasticsearch/test_transport.py b/test_elasticsearch/test_transport.py
index 328325c1..50acb7fa 100644
--- a/test_elasticsearch/test_transport.py
+++ b/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)