# Licensed to Elasticsearch B.V. under one or more contributor # license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright # ownership. Elasticsearch B.V. licenses this file to you under # the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. import base64 import inspect import urllib.parse import warnings from datetime import date, datetime from enum import Enum, auto from functools import wraps from typing import ( TYPE_CHECKING, Any, Awaitable, Callable, Collection, Dict, List, Mapping, Optional, Sequence, Set, Tuple, Type, TypeVar, Union, ) from elastic_transport import ( AsyncTransport, HttpHeaders, NodeConfig, RequestsHttpNode, SniffOptions, Transport, ) from elastic_transport.client_utils import ( DEFAULT, client_meta_version, create_user_agent, parse_cloud_id, url_to_node_config, ) from ..._version import __versionstr__ from ...compat import to_bytes, to_str, warn_stacklevel from ...exceptions import GeneralAvailabilityWarning if TYPE_CHECKING: from ._base import NamespacedClient # parts of URL to be omitted SKIP_IN_PATH: Collection[Any] = (None, "", b"", [], ()) # To be passed to 'client_meta_service' on the Transport CLIENT_META_SERVICE = ("es", client_meta_version(__versionstr__)) # Default User-Agent used by the client USER_AGENT = create_user_agent("elasticsearch-py", __versionstr__) class Stability(Enum): STABLE = auto() BETA = auto() EXPERIMENTAL = auto() _TYPE_HOSTS = Union[ str, Sequence[Union[str, Mapping[str, Union[str, int]], NodeConfig]] ] _TYPE_BODY = Union[bytes, str, Dict[str, Any]] _TYPE_ASYNC_SNIFF_CALLBACK = Callable[ [AsyncTransport, SniffOptions], Awaitable[List[NodeConfig]] ] _TYPE_SYNC_SNIFF_CALLBACK = Callable[[Transport, SniffOptions], List[NodeConfig]] _TRANSPORT_OPTIONS = { "api_key", "http_auth", "request_timeout", "opaque_id", "headers", "ignore", } F = TypeVar("F", bound=Callable[..., Any]) def client_node_configs( hosts: Optional[_TYPE_HOSTS], cloud_id: Optional[str], requests_session_auth: Optional[Any] = None, **kwargs: Any, ) -> List[NodeConfig]: if cloud_id is not None: if hosts is not None: raise ValueError( "The 'cloud_id' and 'hosts' parameters are mutually exclusive" ) node_configs = cloud_id_to_node_configs(cloud_id) else: assert hosts is not None node_configs = hosts_to_node_configs(hosts) # Remove all values which are 'DEFAULT' to avoid overwriting actual defaults. node_options = {k: v for k, v in kwargs.items() if v is not DEFAULT} # Set the 'User-Agent' default header. headers = HttpHeaders(node_options.pop("headers", ())) headers.setdefault("user-agent", USER_AGENT) node_options["headers"] = headers # If a custom Requests AuthBase is passed we set that via '_extras'. if requests_session_auth is not None: node_options.setdefault("_extras", {})[ "requests.session.auth" ] = requests_session_auth def apply_node_options(node_config: NodeConfig) -> NodeConfig: """Needs special handling of headers since .replace() wipes out existing headers""" headers = node_config.headers.copy() # type: ignore[attr-defined] headers_to_add = node_options.pop("headers", ()) if headers_to_add: headers.update(headers_to_add) headers.setdefault("user-agent", USER_AGENT) headers.freeze() node_options["headers"] = headers return node_config.replace(**node_options) return [apply_node_options(node_config) for node_config in node_configs] def hosts_to_node_configs(hosts: _TYPE_HOSTS) -> List[NodeConfig]: """Transforms the many formats of 'hosts' into NodeConfigs""" # To make the logic here simpler we reroute everything to be List[X] if isinstance(hosts, str): return hosts_to_node_configs([hosts]) node_configs: List[NodeConfig] = [] for host in hosts: if isinstance(host, NodeConfig): node_configs.append(host) elif isinstance(host, str): node_configs.append(url_to_node_config(host)) elif isinstance(host, Mapping): node_configs.append(host_mapping_to_node_config(host)) else: raise ValueError( "'hosts' must be a list of URLs, NodeConfigs, or dictionaries" ) return node_configs def host_mapping_to_node_config(host: Mapping[str, Union[str, int]]) -> NodeConfig: """Converts an old-style dictionary host specification to a NodeConfig""" allow_hosts_keys = { "scheme", "host", "port", "path_prefix", } disallowed_keys = set(host.keys()).difference(allow_hosts_keys) if disallowed_keys: bad_keys_used = "', '".join(sorted(disallowed_keys)) allowed_keys = "', '".join(sorted(allow_hosts_keys)) raise ValueError( f"Can't specify the options '{bad_keys_used}' via a " f"dictionary in 'hosts', only '{allowed_keys}' options " "are allowed" ) options = dict(host) return NodeConfig(**options) # type: ignore[arg-type] def cloud_id_to_node_configs(cloud_id: str) -> List[NodeConfig]: """Transforms an Elastic Cloud ID into a NodeConfig""" es_addr = parse_cloud_id(cloud_id).es_address if es_addr is None or not all(es_addr): raise ValueError("Cloud ID missing host and port information for Elasticsearch") host, port = es_addr return [ NodeConfig( scheme="https", host=host, port=port, http_compress=True, ) ] def _base64_auth_header(auth_value: Union[str, List[str], Tuple[str, str]]) -> str: """Takes either a 2-tuple or a base64-encoded string and returns a base64-encoded string to be used as an HTTP authorization header. """ if isinstance(auth_value, (list, tuple)): return base64.b64encode(to_bytes(":".join(auth_value))).decode("ascii") return to_str(auth_value) def _escape(value: Any) -> str: """ Escape a single value of a URL string or a query parameter. If it is a list or tuple, turn it into a comma-separated string first. """ # make sequences into comma-separated stings if isinstance(value, (list, tuple)): value = ",".join([_escape(item) for item in value]) # dates and datetimes into isoformat elif isinstance(value, (date, datetime)): value = value.isoformat() # make bools into true/false strings elif isinstance(value, bool): value = str(value).lower() elif isinstance(value, bytes): return value.decode("utf-8", "surrogatepass") if not isinstance(value, str): return str(value) return value def _quote(value: Any) -> str: return urllib.parse.quote(_escape(value), ",*") def _quote_query(query: Mapping[str, Any]) -> str: return "&".join([f"{k}={_quote(v)}" for k, v in query.items()]) def _merge_kwargs_no_duplicates(kwargs: Dict[str, Any], values: Dict[str, Any]) -> None: for key, val in values.items(): if key in kwargs: raise ValueError( f"Received multiple values for '{key}', specify parameters " "directly instead of using 'params'" ) kwargs[key] = val def _merge_body_fields_no_duplicates( body: _TYPE_BODY, kwargs: Dict[str, Any], body_fields: Tuple[str, ...] ) -> bool: mixed_body_and_params = False for key in list(kwargs.keys()): if key in body_fields: if isinstance(body, (str, bytes)): raise ValueError( "Couldn't merge 'body' with other parameters as it wasn't a mapping." ) if key in body: raise ValueError( f"Received multiple values for '{key}', specify parameters " "using either body or parameters, not both." ) warnings.warn( f"Received '{key}' via a specific parameter in the presence of a " "'body' parameter, which is deprecated and will be removed in a future " "version. Instead, use only 'body' or only specific parameters.", category=DeprecationWarning, stacklevel=warn_stacklevel(), ) body[key] = kwargs.pop(key) mixed_body_and_params = True return mixed_body_and_params def _rewrite_parameters( body_name: Optional[str] = None, body_fields: Optional[Tuple[str, ...]] = None, parameter_aliases: Optional[Dict[str, str]] = None, ignore_deprecated_options: Optional[Set[str]] = None, ) -> Callable[[F], F]: def wrapper(api: F) -> F: @wraps(api) def wrapped(*args: Any, **kwargs: Any) -> Any: # Let's give a nicer error message when users pass positional arguments. if len(args) >= 2: raise TypeError( "Positional arguments can't be used with Elasticsearch API methods. " "Instead only use keyword arguments." ) # We merge 'params' first as transport options can be specified using params. if "params" in kwargs and ( not ignore_deprecated_options or "params" not in ignore_deprecated_options ): params = kwargs.pop("params") if params: if not hasattr(params, "items"): raise ValueError( "Couldn't merge 'params' with other parameters as it wasn't a mapping. " "Instead of using 'params' use individual API parameters" ) warnings.warn( "The 'params' parameter is deprecated and will be removed " "in a future version. Instead use individual parameters.", category=DeprecationWarning, stacklevel=warn_stacklevel(), ) _merge_kwargs_no_duplicates(kwargs, params) maybe_transport_options = _TRANSPORT_OPTIONS.intersection(kwargs) if maybe_transport_options: transport_options = {} for option in maybe_transport_options: if ( ignore_deprecated_options and option in ignore_deprecated_options ): continue try: option_rename = option if option == "ignore": option_rename = "ignore_status" transport_options[option_rename] = kwargs.pop(option) except KeyError: pass if transport_options: warnings.warn( "Passing transport options in the API method is deprecated. Use 'Elasticsearch.options()' instead.", category=DeprecationWarning, stacklevel=warn_stacklevel(), ) client = args[0] # Namespaced clients need to unwrapped. namespaced_client: Optional[Type["NamespacedClient"]] = None if hasattr(client, "_client"): namespaced_client = type(client) client = client._client client = client.options(**transport_options) # Re-wrap the client if we unwrapped due to being namespaced. if namespaced_client is not None: client = namespaced_client(client) args = (client,) + args[1:] if "body" in kwargs and ( not ignore_deprecated_options or "body" not in ignore_deprecated_options ): body: Optional[_TYPE_BODY] = kwargs.pop("body") mixed_body_and_params = False if body is not None: if body_name: if body_name in kwargs: raise TypeError( f"Can't use '{body_name}' and 'body' parameters together because '{body_name}' " "is an alias for 'body'. Instead you should only use the " f"'{body_name}' parameter. See https://github.com/elastic/elasticsearch-py/" "issues/1698 for more information" ) kwargs[body_name] = body elif body_fields is not None: mixed_body_and_params = _merge_body_fields_no_duplicates( body, kwargs, body_fields ) kwargs["body"] = body if parameter_aliases and not isinstance(body, (str, bytes)): for alias, rename_to in parameter_aliases.items(): if rename_to in body: body[alias] = body.pop(rename_to) # If body and params are mixed, the alias may come from a param, # in which case the warning below will not make sense. if not mixed_body_and_params: warnings.warn( f"Using '{rename_to}' alias in 'body' is deprecated and will be removed " f"in a future version of elasticsearch-py. Use '{alias}' directly instead. " "See https://github.com/elastic/elasticsearch-py/issues/1698 for more information", category=DeprecationWarning, stacklevel=2, ) if parameter_aliases: for alias, rename_to in parameter_aliases.items(): try: kwargs[rename_to] = kwargs.pop(alias) except KeyError: pass return api(*args, **kwargs) return wrapped # type: ignore[return-value] 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) 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: from requests.auth import AuthBase return isinstance(http_auth, AuthBase) except ImportError: pass return False def is_requests_node_class(node_class: Any) -> bool: """Detect if 'RequestsHttpNode' would be used given the setting of 'node_class'""" return ( node_class is not None and node_class is not DEFAULT and ( node_class == "requests" or ( inspect.isclass(node_class) and issubclass(node_class, RequestsHttpNode) ) ) )