Add back handling of the 'Warning' HTTP header

Co-authored-by: Seth Michael Larson <seth.larson@elastic.co>

Author: github-actions[bot]

elasticsearch/_async/client/_base.py

@@ -15,18 +15,27 @@
 #  specific language governing permissions and limitations
 #  under the License.
 
-from typing import Any, Collection, Mapping, Optional, Tuple, TypeVar, Union
+import re
+import warnings
+from typing import Any, Collection, Iterable, Mapping, Optional, Tuple, TypeVar, Union
 
 from elastic_transport import AsyncTransport, HttpHeaders
 from elastic_transport.client_utils import DEFAULT, DefaultType, resolve_default
 
-from ...compat import urlencode
-from ...exceptions import HTTP_EXCEPTIONS, ApiError, UnsupportedProductError
+from ...compat import urlencode, warn_stacklevel
+from ...exceptions import (
+    HTTP_EXCEPTIONS,
+    ApiError,
+    ElasticsearchWarning,
+    UnsupportedProductError,
+)
 from .utils import _base64_auth_header
 
 SelfType = TypeVar("SelfType", bound="BaseClient")
 SelfNamespacedType = TypeVar("SelfNamespacedType", bound="NamespacedClient")
 
+_WARNING_RE = re.compile(r"\"([^\"]*)\"")
+
 
 def resolve_auth_headers(
     headers: Optional[Mapping[str, str]],
@@ -153,6 +162,19 @@ async def _perform_request(
                 body=response,
             )
 
+        # 'Warning' headers should be reraised as 'ElasticsearchWarning'
+        warning_header = (meta.headers.get("warning") or "").strip()
+        if warning_header:
+            warning_messages: Iterable[str] = _WARNING_RE.findall(warning_header) or (
+                warning_header,
+            )
+            for warning_message in warning_messages:
+                warnings.warn(
+                    warning_message,
+                    category=ElasticsearchWarning,
+                    stacklevel=warn_stacklevel(),
+                )
+
         return response
 
     def options(

elasticsearch/_sync/client/_base.py

@@ -15,18 +15,27 @@
 #  specific language governing permissions and limitations
 #  under the License.
 
+import re
+import warnings
 from typing import Any, Collection, Mapping, Optional, Tuple, TypeVar, Union
 
 from elastic_transport import HttpHeaders, Transport
 from elastic_transport.client_utils import DEFAULT, DefaultType, resolve_default
 
-from ...compat import urlencode
-from ...exceptions import HTTP_EXCEPTIONS, ApiError, UnsupportedProductError
+from ...compat import urlencode, warn_stacklevel
+from ...exceptions import (
+    HTTP_EXCEPTIONS,
+    ApiError,
+    ElasticsearchWarning,
+    UnsupportedProductError,
+)
 from .utils import _base64_auth_header
 
 SelfType = TypeVar("SelfType", bound="BaseClient")
 SelfNamespacedType = TypeVar("SelfNamespacedType", bound="NamespacedClient")
 
+_WARNING_RE = re.compile(r"\"([^\"]*)\"")
+
 
 def resolve_auth_headers(
     headers: Optional[Mapping[str, str]],
@@ -153,6 +162,18 @@ def _perform_request(
                 body=response,
             )
 
+        # 'Warning' headers should be reraised as 'ElasticsearchWarning'
+        warning_header = (meta.headers.get("warning") or "").strip()
+        if warning_header:
+            for warning_message in _WARNING_RE.findall(warning_header) or (
+                warning_header,
+            ):
+                warnings.warn(
+                    warning_message,
+                    category=ElasticsearchWarning,
+                    stacklevel=warn_stacklevel(),
+                )
+
         return response
 
     def options(

elasticsearch/compat.py

@@ -15,11 +15,12 @@
 #  specific language governing permissions and limitations
 #  under the License.
 
-from queue import Queue
+import inspect
+import sys
+from pathlib import Path
 from typing import Mapping, Tuple, Type, Union
-from urllib.parse import quote, quote_plus, unquote
+from urllib.parse import quote
 from urllib.parse import urlencode as _urlencode
-from urllib.parse import urlparse
 
 from elastic_transport.client_utils import percent_encode
 
@@ -42,30 +43,47 @@ def to_bytes(x: Union[str, bytes], encoding: str = "ascii") -> bytes:
     return x
 
 
-try:
-    from collections.abc import Mapping
-except ImportError:
-    from collections import Mapping
-
-
-reraise_exceptions: Tuple[Type[Exception], ...] = (RecursionError,)
-
-try:
-    import asyncio
-
-    reraise_exceptions += (asyncio.CancelledError,)
-except (ImportError, AttributeError):
-    pass
+def warn_stacklevel() -> int:
+    """Dynamically determine warning stacklevel for warnings based on the call stack"""
+    try:
+        # Grab the root module from the current module '__name__'
+        module_name = __name__.partition(".")[0]
+        module_path = Path(sys.modules[module_name].__file__)
+
+        # If the module is a folder we're looking at
+        # subdirectories, otherwise we're looking for
+        # an exact match.
+        module_is_folder = module_path.name == "__init__.py"
+        if module_is_folder:
+            module_path = module_path.parent
+
+        # Look through frames until we find a file that
+        # isn't a part of our module, then return that stacklevel.
+        for level, frame in enumerate(inspect.stack()):
+            # Garbage collecting frames
+            frame_filename = Path(frame.filename)
+            del frame
+
+            if (
+                # If the module is a folder we look at subdirectory
+                module_is_folder
+                and module_path not in frame_filename.parents
+            ) or (
+                # Otherwise we're looking for an exact match.
+                not module_is_folder
+                and module_path != frame_filename
+            ):
+                return level
+    except KeyError:
+        pass
+    return 0
 
 
 __all__ = [
     "string_types",
-    "reraise_exceptions",
-    "quote_plus",
+    "to_str",
+    "to_bytes",
     "quote",
     "urlencode",
-    "unquote",
-    "urlparse",
-    "Queue",
-    "Mapping",
+    "warn_stacklevel",
 ]

elasticsearch/helpers/actions.py

@@ -18,8 +18,10 @@
 import logging
 import time
 from operator import methodcaller
+from queue import Queue
+from typing import Mapping
 
-from ..compat import Mapping, Queue, string_types, to_bytes
+from ..compat import string_types, to_bytes
 from ..exceptions import NotFoundError, TransportError
 from .errors import BulkIndexError, ScanError
 

test_elasticsearch/test_async/test_transport.py

@@ -21,6 +21,7 @@
 import asyncio
 import json
 import re
+import warnings
 
 import pytest
 from elastic_transport import ApiResponseMeta, BaseAsyncNode, HttpHeaders, NodeConfig
@@ -30,6 +31,7 @@
 from elasticsearch import AsyncElasticsearch
 from elasticsearch.exceptions import (
     ConnectionError,
+    ElasticsearchWarning,
     TransportError,
     UnsupportedProductError,
 )
@@ -622,3 +624,34 @@ async def test_transport_error_raised_before_product_error(status):
     calls = client.transport.node_pool.get().calls
     assert len(calls) == 1
     assert calls[0][0] == ("GET", "/")
+
+
+@pytest.mark.parametrize(
+    "headers",
+    [
+        {
+            "Warning": '299 Elasticsearch-8.0.0-SNAPSHOT-ad975cacd240b3329e160673c432e768dcd7899a "[xpack.monitoring.history.duration] setting was deprecated in Elasticsearch and will be removed in a future release! See the breaking changes documentation for the next major version."',
+            "X-elastic-product": "Elasticsearch",
+        },
+        {
+            "Warning": '299 Elasticsearch-8.0.0-SNAPSHOT-ad975cacd240b3329e160673c432e768dcd7899a "[xpack.monitoring.history.duration] setting was deprecated in Elasticsearch and will be removed in a future release! See the breaking changes documentation for the next major version.", 299 Elasticsearch-8.0.0-SNAPSHOT-ad975cacd240b3329e160673c432e768dcd7899a "[xpack.monitoring.history.duration2] setting was deprecated in Elasticsearch and will be removed in a future release! See the breaking changes documentation for the next major version."',
+            "X-elastic-product": "Elasticsearch",
+        },
+    ],
+)
+async def test_warning_header(headers):
+    client = AsyncElasticsearch(
+        [NodeConfig("http", "localhost", 9200, _extras={"headers": headers})],
+        meta_header=False,
+        node_class=DummyNode,
+    )
+
+    with warnings.catch_warnings(record=True) as w:
+        await client.info()
+
+    assert len(w) == headers["Warning"].count("299")
+    assert w[0].category == ElasticsearchWarning
+    assert (
+        str(w[0].message)
+        == "[xpack.monitoring.history.duration] setting was deprecated in Elasticsearch and will be removed in a future release! See the breaking changes documentation for the next major version."
+    )

test_elasticsearch/test_server/test_rest_api_spec.py

@@ -275,13 +275,13 @@ def run_catch(self, catch, exception):
 
         assert isinstance(exception, ApiError)
         if catch in CATCH_CODES:
-            assert CATCH_CODES[catch] == exception.status
+            assert CATCH_CODES[catch] == exception.meta.status
         elif catch[0] == "/" and catch[-1] == "/":
             assert (
-                re.search(catch[1:-1], str(exception.message)),
-                f"{catch} not in {str(exception.message)!r}",
+                re.search(catch[1:-1], str(exception.body)),
+                f"{catch} not in {str(exception.body)!r}",
             ) is not None
-        self.last_response = exception.message
+        self.last_response = exception.body
 
     def run_skip(self, skip):
         global IMPLEMENTED_FEATURES

test_elasticsearch/test_transport.py

@@ -21,6 +21,7 @@
 import json
 import re
 import time
+import warnings
 
 import pytest
 from elastic_transport import ApiResponseMeta, BaseNode, HttpHeaders, NodeConfig
@@ -30,6 +31,7 @@
 from elasticsearch import Elasticsearch
 from elasticsearch.exceptions import (
     ConnectionError,
+    ElasticsearchWarning,
     TransportError,
     UnsupportedProductError,
 )
@@ -504,3 +506,34 @@ def test_transport_error_raised_before_product_error(status):
     calls = client.transport.node_pool.get().calls
     assert len(calls) == 1
     assert calls[0][0] == ("GET", "/")
+
+
+@pytest.mark.parametrize(
+    "headers",
+    [
+        {
+            "Warning": '299 Elasticsearch-8.0.0-SNAPSHOT-ad975cacd240b3329e160673c432e768dcd7899a "[xpack.monitoring.history.duration] setting was deprecated in Elasticsearch and will be removed in a future release! See the breaking changes documentation for the next major version."',
+            "X-elastic-product": "Elasticsearch",
+        },
+        {
+            "Warning": '299 Elasticsearch-8.0.0-SNAPSHOT-ad975cacd240b3329e160673c432e768dcd7899a "[xpack.monitoring.history.duration] setting was deprecated in Elasticsearch and will be removed in a future release! See the breaking changes documentation for the next major version.", 299 Elasticsearch-8.0.0-SNAPSHOT-ad975cacd240b3329e160673c432e768dcd7899a "[xpack.monitoring.history.duration2] setting was deprecated in Elasticsearch and will be removed in a future release! See the breaking changes documentation for the next major version."',
+            "X-elastic-product": "Elasticsearch",
+        },
+    ],
+)
+def test_warning_header(headers):
+    client = Elasticsearch(
+        [NodeConfig("http", "localhost", 9200, _extras={"headers": headers})],
+        meta_header=False,
+        node_class=DummyNode,
+    )
+
+    with warnings.catch_warnings(record=True) as w:
+        client.info()
+
+    assert len(w) == headers["Warning"].count("299")
+    assert w[0].category == ElasticsearchWarning
+    assert (
+        str(w[0].message)
+        == "[xpack.monitoring.history.duration] setting was deprecated in Elasticsearch and will be removed in a future release! See the breaking changes documentation for the next major version."
+    )