Add API warnings infrastructure

Author: pquentin

elasticsearch/_async/client/utils.py

@@ -20,10 +20,12 @@
     _TYPE_HOSTS,
     CLIENT_META_SERVICE,
     SKIP_IN_PATH,
+    Stability,
     _base64_auth_header,
     _quote,
     _quote_query,
     _rewrite_parameters,
+    _stability_warning,
     client_node_configs,
     is_requests_http_auth,
     is_requests_node_class,
@@ -37,8 +39,10 @@
     "_quote_query",
     "_TYPE_HOSTS",
     "SKIP_IN_PATH",
+    "Stability",
     "client_node_configs",
     "_rewrite_parameters",
+    "_stability_warning",
     "is_requests_http_auth",
     "is_requests_node_class",
 ]

elasticsearch/_sync/client/utils.py

@@ -20,6 +20,7 @@
 import urllib.parse
 import warnings
 from datetime import date, datetime
+from enum import Enum, auto
 from functools import wraps
 from typing import (
     TYPE_CHECKING,
@@ -55,6 +56,8 @@
     url_to_node_config,
 )
 
+from elasticsearch.exceptions import GeneralAvailabilityWarning
+
 from ..._version import __versionstr__
 from ...compat import to_bytes, to_str, warn_stacklevel
 
@@ -93,6 +96,13 @@
 F = TypeVar("F", bound=Callable[..., Any])
 
 
+class Stability(Enum):
+    STABLE = auto()
+    BETA = auto()
+    EXPERIMENTAL = auto()
+    DEPRECATED = auto()
+
+
 def client_node_configs(
     hosts: Optional[_TYPE_HOSTS],
     cloud_id: Optional[str],
@@ -450,6 +460,42 @@ def wrapped(*args: Any, **kwargs: Any) -> Any:
     return wrapper
 
 
+def _stability_warning(
+    stability: Stability,
+    version: Optional[str] = None,
+    message: Optional[str] = None,
+) -> Callable[[F], F]:
+    def wrapper(api: F) -> F:
+        @wraps(api)
+        def wrapped(*args: Any, **kwargs: Any) -> Any:
+            if stability == Stability.BETA:
+                warnings.warn(
+                    "This API is in beta and is subject to change. "
+                    "The design and code is less mature than official GA features and is being provided as-is with no warranties. "
+                    "Beta features are not subject to the support SLA of official GA features.",
+                    category=GeneralAvailabilityWarning,
+                    stacklevel=warn_stacklevel(),
+                )
+            elif stability == Stability.EXPERIMENTAL:
+                warnings.warn(
+                    "This API is in technical preview and may be changed or removed in a future release. "
+                    "Elastic will work to fix any issues, but features in technical preview are not subject to the support SLA of official GA features.",
+                    category=GeneralAvailabilityWarning,
+                    stacklevel=warn_stacklevel(),
+                )
+                return api(*args, **kwargs)
+            elif stability == Stability.DEPRECATED:
+                warnings.warn(
+                    f"This API was deprecated in Elasticsearch {version}. {message}",
+                    category=DeprecationWarning,
+                    stacklevel=warn_stacklevel(),
+                )
+
+        return wrapped  # type: ignore[return-value]
+
+    return wrapper
+
+
 def is_requests_http_auth(http_auth: Any) -> bool:
     """Detect if an http_auth value is a custom Requests auth object"""
     try:

elasticsearch/exceptions.py

@@ -115,6 +115,10 @@ class ElasticsearchWarning(TransportWarning):
     """
 
 
+class GeneralAvailabilityWarning(TransportWarning):
+    """ "Warning that is raised when a feature is not yet GA."""
+
+
 # Aliases for backwards compatibility
 ElasticsearchDeprecationWarning = ElasticsearchWarning
 RequestError = BadRequestError

test_elasticsearch/test_client/test_utils.py

@@ -16,7 +16,10 @@
 #  under the License.
 
 
-from elasticsearch._sync.client.utils import _quote
+import warnings
+
+from elasticsearch._sync.client.utils import Stability, _quote, _stability_warning
+from elasticsearch.exceptions import GeneralAvailabilityWarning
 
 
 def test_handles_ascii():
@@ -36,3 +39,62 @@ def test_handles_unicode():
 def test_handles_unicode2():
     string = "中*文,"
     assert "%E4%B8%AD*%E6%96%87," == _quote(string)
+
+
+class TestStabilityWarning:
+    def test_default(self):
+
+        @_stability_warning(stability=Stability.STABLE)
+        def func_default(*args, **kwargs):
+            pass
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("error")
+            func_default()
+
+    def test_beta(self, recwarn):
+
+        @_stability_warning(stability=Stability.BETA)
+        def func_beta(*args, **kwargs):
+            pass
+
+        func_beta()
+
+        assert len(recwarn) == 1
+        user_warning = recwarn.pop(GeneralAvailabilityWarning)
+        assert user_warning.category == GeneralAvailabilityWarning
+        assert user_warning.message.args[0].startswith(
+            "This API is in beta and is subject to change."
+        )
+
+    def test_experimental(self, recwarn):
+
+        @_stability_warning(stability=Stability.EXPERIMENTAL)
+        def func_experimental(*args, **kwargs):
+            pass
+
+        func_experimental()
+
+        assert len(recwarn) == 1
+        user_warning = recwarn.pop(GeneralAvailabilityWarning)
+        assert user_warning.category == GeneralAvailabilityWarning
+        assert user_warning.message.args[0].startswith(
+            "This API is in technical preview and may be changed or removed in a future release."
+        )
+
+    def test_deprecated(self, recwarn):
+
+        @_stability_warning(
+            stability=Stability.DEPRECATED, version="8.4.0", message="Use bar instead."
+        )
+        def func_deprecated(*args, **kwargs):
+            pass
+
+        func_deprecated()
+
+        assert len(recwarn) == 1
+        user_warning = recwarn.pop(DeprecationWarning)
+        assert user_warning.category == DeprecationWarning
+        assert user_warning.message.args[0] == (
+            "This API was deprecated in Elasticsearch 8.4.0. Use bar instead."
+        )