[7.x] Add support for the HTTP API compatibility header

Co-authored-by: Seth Michael Larson <[email protected]>

docs/sphinx/connection.rst

@@ -60,7 +60,29 @@ To create an `SSLContext` object you only need to use one of cafile, capath or c
 * `capath` is the directory of a collection of CA's
 * `cadata` is either an ASCII string of one or more PEM-encoded certificates or a bytes-like object of DER-encoded certificates.
 
-Please note that the use of SSLContext is only available for Urllib3.
+Please note that the use of SSLContext is only available for urllib3.
 
 .. autoclass:: Urllib3HttpConnection
    :members:
+
+
+API Compatibility HTTP Header
+-----------------------------
+
+The Python client can be configured to emit an HTTP header
+``Accept: application/vnd.elasticsearch+json; compatible-with=7``
+which signals to Elasticsearch that the client is requesting
+``7.x`` version of request and response bodies. This allows for
+upgrading from 7.x to 8.x version of Elasticsearch without upgrading
+everything at once. Elasticsearch should be upgraded first after
+the compatibility header is configured and clients should be upgraded
+second.
+
+ .. code-block:: python
+
+    from elasticsearch import Elasticsearch
+
+    client = Elasticsearch("http://...", headers={"accept": "application/vnd.elasticsearch+json; compatible-with=7"})
+
+If you'd like to have the client emit the header without configuring ``headers`` you
+can use the environment variable ``ELASTIC_CLIENT_APIVERSIONING=1``.

elasticsearch/connection/base.py

@@ -19,6 +19,7 @@ import binascii
 import gzip
 import io
 import logging
+import os
 import re
 import warnings
 from platform import python_version
@@ -28,7 +29,7 @@ try:
 except ImportError:
     import json
 
-from .. import __versionstr__
+from .. import __version__, __versionstr__
 from ..exceptions import (
     HTTP_EXCEPTIONS,
     ElasticsearchWarning,
@@ -121,6 +122,13 @@ class Connection(object):
         if opaque_id:
             self.headers["x-opaque-id"] = opaque_id
 
+        if os.getenv("ELASTIC_CLIENT_APIVERSIONING") == "1":
+            self.headers.setdefault(
+                "accept",
+                "application/vnd.elasticsearch+json;compatible-with=%s"
+                % (str(__version__[0]),),
+            )
+
         self.headers.setdefault("content-type", "application/json")
         self.headers.setdefault("user-agent", self._get_default_user_agent())
 
@@ -248,7 +256,7 @@ class Connection(object):
     def log_request_success(
         self, method, full_url, path, body, status_code, response, duration
     ):
-        """ Log a successful API call.  """
+        """Log a successful API call."""
         #  TODO: optionally pass in params instead of full_url and do urlencode only when needed
 
         # body has already been serialized to utf-8, deserialize it for logging
@@ -278,7 +286,7 @@ class Connection(object):
         response=None,
         exception=None,
     ):
-        """ Log an unsuccessful API call.  """
+        """Log an unsuccessful API call."""
         # do not log 404s on HEAD requests
         if method == "HEAD" and status_code == 404:
             return
@@ -307,7 +315,7 @@ class Connection(object):
             logger.debug("< %s", response)
 
     def _raise_error(self, status_code, raw_data):
-        """ Locate appropriate exception and raise it. """
+        """Locate appropriate exception and raise it."""
         error_message = raw_data
         additional_info = None
         try:

elasticsearch/exceptions.py

@@ -68,7 +68,7 @@ class TransportError(ElasticsearchException):
 
     @property
     def error(self):
-        """ A string error message. """
+        """A string error message."""
         return self.args[1]
 
     @property
@@ -120,11 +120,11 @@ class ConnectionError(TransportError):
 
 
 class SSLError(ConnectionError):
-    """ Error raised when encountering SSL errors. """
+    """Error raised when encountering SSL errors."""
 
 
 class ConnectionTimeout(ConnectionError):
-    """ A network timeout. Doesn't cause a node retry by default. """
+    """A network timeout. Doesn't cause a node retry by default."""
 
     def __str__(self):
         return "ConnectionTimeout caused by - %s(%s)" % (
@@ -134,23 +134,23 @@ class ConnectionTimeout(ConnectionError):
 
 
 class NotFoundError(TransportError):
-    """ Exception representing a 404 status code. """
+    """Exception representing a 404 status code."""
 
 
 class ConflictError(TransportError):
-    """ Exception representing a 409 status code. """
+    """Exception representing a 409 status code."""
 
 
 class RequestError(TransportError):
-    """ Exception representing a 400 status code. """
+    """Exception representing a 400 status code."""
 
 
 class AuthenticationException(TransportError):
-    """ Exception representing a 401 status code. """
+    """Exception representing a 401 status code."""
 
 
 class AuthorizationException(TransportError):
-    """ Exception representing a 403 status code. """
+    """Exception representing a 403 status code."""
 
 
 class ElasticsearchWarning(Warning):

elasticsearch/helpers/errors.py

@@ -21,7 +21,7 @@ from ..exceptions import ElasticsearchException
 class BulkIndexError(ElasticsearchException):
     @property
     def errors(self):
-        """ List of errors from execution of the last chunk. """
+        """List of errors from execution of the last chunk."""
         return self.args[1]
 
 

elasticsearch/serializer.py

@@ -154,8 +154,12 @@ class Deserializer(object):
         if not mimetype:
             deserializer = self.default
         else:
-            # split out charset
+            # split out 'charset' and 'compatible-width' options
-            mimetype, _, _ = mimetype.partition(";")
+            mimetype = mimetype.partition(";")[0].strip()
+            # Treat 'application/vnd.elasticsearch+json'
+            # as application/json for compatibility.
+            if mimetype == "application/vnd.elasticsearch+json":
+                mimetype = "application/json"
             try:
                 deserializer = self.serializers[mimetype]
             except KeyError:

test_elasticsearch/test_async/test_server/test_rest_api_spec.py

@@ -75,7 +75,7 @@ class AsyncYamlRunner(YamlRunner):
             await self.teardown()
 
     async def run_code(self, test):
-        """ Execute an instruction based on it's type. """
+        """Execute an instruction based on it's type."""
         print(test)
         for action in test:
             assert len(action) == 1

test_elasticsearch/test_connection.py

@@ -18,6 +18,7 @@
 
 import gzip
 import io
+import os
 import re
 import ssl
 import warnings
@@ -171,6 +172,26 @@ class TestBaseConnection(TestCase):
             Connection(meta_header=1)
         assert str(e.value) == "meta_header must be of type bool"
 
+    def test_compatibility_accept_header(self):
+        try:
+            conn = Connection()
+            assert "accept" not in conn.headers
+
+            os.environ["ELASTIC_CLIENT_APIVERSIONING"] = "0"
+
+            conn = Connection()
+            assert "accept" not in conn.headers
+
+            os.environ["ELASTIC_CLIENT_APIVERSIONING"] = "1"
+
+            conn = Connection()
+            assert (
+                conn.headers["accept"]
+                == "application/vnd.elasticsearch+json;compatible-with=8"
+            )
+        finally:
+            os.environ.pop("ELASTIC_CLIENT_APIVERSIONING")
+
 
 class TestUrllib3Connection(TestCase):
     def _get_mock_connection(self, connection_params={}, response_body=b"{}"):

test_elasticsearch/test_serializer.py

@@ -185,6 +185,17 @@ class TestDeserializer(TestCase):
             self.de.loads('{"some":"data"}', "text/plain; charset=whatever"),
         )
 
+    def test_deserialize_compatibility_header(self):
+        for content_type in (
+            "application/vnd.elasticsearch+json;compatible-with=7",
+            "application/vnd.elasticsearch+json; compatible-with=7",
+            "application/vnd.elasticsearch+json;compatible-with=8",
+            "application/vnd.elasticsearch+json; compatible-with=8",
+        ):
+            self.assertEqual(
+                {"some": "data"}, self.de.loads('{"some":"data"}', content_type)
+            )
+
     def test_raises_serialization_error_on_unknown_mimetype(self):
         self.assertRaises(SerializationError, self.de.loads, "{}", "text/html")
 

test_elasticsearch/test_server/test_rest_api_spec.py

@@ -138,7 +138,7 @@ class YamlRunner:
             self.teardown()
 
     def run_code(self, test):
-        """ Execute an instruction based on it's type. """
+        """Execute an instruction based on it's type."""
         print(test)
         for action in test:
             assert len(action) == 1