From 073db90ae855aed692b3efaa4965a1d3b6b7f692 Mon Sep 17 00:00:00 2001 From: Joshua Carp Date: Wed, 12 Jul 2017 22:34:17 -0400 Subject: [PATCH] Set application/x-ndjson content type on bulk requests. Resolves #609 --- elasticsearch/client/__init__.py | 3 ++- elasticsearch/connection/base.py | 2 -- elasticsearch/connection/http_requests.py | 4 ++-- elasticsearch/connection/http_urllib3.py | 4 +++- elasticsearch/transport.py | 8 +++++--- test_elasticsearch/test_transport.py | 2 +- 6 files changed, 13 insertions(+), 10 deletions(-) diff --git a/elasticsearch/client/__init__.py b/elasticsearch/client/__init__.py index 59858549..22f2bd35 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') diff --git a/elasticsearch/connection/base.py b/elasticsearch/connection/base.py index 3ea18fd5..44fbacee 100644 --- a/elasticsearch/connection/base.py +++ b/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) - - 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..7cd754b0 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,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') diff --git a/elasticsearch/transport.py b/elasticsearch/transport.py index dc8cd891..ff5a6a17 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 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: 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)