Clean up custom headers.

elasticsearch/client/__init__.py

@@ -1128,7 +1128,7 @@ class Elasticsearch(object):
             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),
-            headers={'Content-Type': 'application/x-ndjson'})
+            headers={'content-type': 'application/x-ndjson'})
 
     @query_params('max_concurrent_searches', 'pre_filter_shard_size',
         'search_type', 'typed_keys')
@@ -1160,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',
@@ -1364,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')
@@ -1388,3 +1390,4 @@ class Elasticsearch(object):
         """
         return self.transport.perform_request('GET', _make_path(index,
             '_field_caps'), params=params, body=body)
+

elasticsearch/connection/base.py

@@ -123,3 +123,5 @@ 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_urllib3.py

@@ -111,8 +111,9 @@ class Urllib3HttpConnection(Connection):
             if not isinstance(method, str):
                 method = method.encode('utf-8')
 
-            request_headers = dict(self.headers)
-            request_headers.update(headers or {})
+            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

@@ -270,7 +270,7 @@ 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
+            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
@@ -294,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

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)