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: 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)