# 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. from __future__ import unicode_literals import base64 import os import warnings import weakref from datetime import date, datetime from functools import wraps from .._version import __versionstr__ from ..compat import PY2, quote, string_types, to_bytes, to_str, unquote, urlparse # parts of URL to be omitted SKIP_IN_PATH = (None, "", b"", [], ()) # Switch to this mimetype if 'ELASTIC_CLIENT_APIVERSIONING=1/true' _COMPATIBILITY_MIMETYPE = "application/vnd.elasticsearch+json;compatible-with=%s" % ( __versionstr__.partition(".")[0] ) def _normalize_hosts(hosts): """ Helper function to transform hosts argument to :class:`~elasticsearch.Elasticsearch` to a list of dicts. """ # if hosts are empty, just defer to defaults down the line if hosts is None: return [{}] # passed in just one string if isinstance(hosts, string_types): hosts = [hosts] out = [] # normalize hosts to dicts for host in hosts: if isinstance(host, string_types): if "://" not in host: host = "//%s" % host parsed_url = urlparse(host) h = {"host": parsed_url.hostname} if parsed_url.port: h["port"] = parsed_url.port if parsed_url.scheme == "https": h["port"] = parsed_url.port or 443 h["use_ssl"] = True if parsed_url.username or parsed_url.password: h["http_auth"] = "%s:%s" % ( unquote(parsed_url.username), unquote(parsed_url.password), ) if parsed_url.path and parsed_url.path != "/": h["url_prefix"] = parsed_url.path out.append(h) else: out.append(host) return out def _escape(value): """ 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(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() # don't decode bytestrings elif isinstance(value, bytes): return value # encode strings to utf-8 if isinstance(value, string_types): if PY2 and isinstance(value, unicode): # noqa: F821 return value.encode("utf-8") if not PY2 and isinstance(value, str): return value.encode("utf-8") return str(value) def _make_path(*parts): """ Create a URL string from parts, omit all `None` values and empty strings. Convert lists and tuples to comma separated values. """ # TODO: maybe only allow some parts to be lists/tuples ? return "/" + "/".join( # preserve ',' and '*' in url for nicer URLs in logs quote(_escape(p), b",*") for p in parts if p not in SKIP_IN_PATH ) # parameters that apply to all methods GLOBAL_PARAMS = ("pretty", "human", "error_trace", "format", "filter_path") def query_params(*es_query_params, **kwargs): """ Decorator that pops all accepted parameters from method's kwargs and puts them in the params argument. """ request_mimetypes = kwargs.pop("request_mimetypes", []) response_mimetypes = kwargs.pop("response_mimetypes", []) default_content_type = "".join(request_mimetypes[:1]) default_accept = ",".join(response_mimetypes) def compat_mimetype(mimetypes): return [ _COMPATIBILITY_MIMETYPE if mimetype in ( "application/json", "application/x-ndjson", "application/vnd.mapbox-vector-tile", ) else mimetype for mimetype in mimetypes ] compat_content_type = "".join(compat_mimetype(request_mimetypes)[:1]) compat_accept = ",".join(compat_mimetype(response_mimetypes)) body_params = kwargs.pop("body_params", None) body_only_params = set(body_params or ()) - set(es_query_params) body_name = kwargs.pop("body_name", None) body_required = kwargs.pop("body_required", False) type_possible_in_params = "type" in es_query_params # There should be no APIs defined with both 'body_params' and a named body. assert not (body_name and body_params) # 'body_required' implies there's no named body and that body_params are defined. assert not (body_name and body_required) assert not body_required or body_params def _wrapper(func): @wraps(func) def _wrapped(*args, **kwargs): params = (kwargs.pop("params", None) or {}).copy() headers = { k.lower(): v for k, v in (kwargs.pop("headers", None) or {}).copy().items() } if "opaque_id" in kwargs: headers["x-opaque-id"] = kwargs.pop("opaque_id") # Detect compatibility mode and set the 'Accept' and 'Content-Type' # headers to the compatibility mimetype if detected. try: if os.environ["ELASTIC_CLIENT_APIVERSIONING"] not in ("true", "1"): raise KeyError # Unset is the same as env var not being 'true' or '1' accept = compat_accept content_type = compat_content_type except KeyError: accept = default_accept content_type = default_content_type # Set the mimetype headers for the request if accept: headers.setdefault("accept", accept) if content_type: headers.setdefault("content-type", content_type) http_auth = kwargs.pop("http_auth", None) api_key = kwargs.pop("api_key", None) # Detect when we should absorb body parameters into 'body' # We only do this when there's no 'body' parameter, no # positional arguments, and at least one parameter we can # serialize in the body. using_body_kwarg = kwargs.get("body", None) is not None using_positional_args = args and len(args) > 1 # The 'doc_type' parameter is deprecated in the query # string. This was generated and missed in 7.x so to # push users to use 'type' instead of 'doc_type' in 8.x # we deprecate it here. if type_possible_in_params: doc_type_in_params = params and "doc_type" in params doc_type_in_kwargs = "doc_type" in kwargs if doc_type_in_params or doc_type_in_kwargs: warnings.warn( "The 'doc_type' parameter is deprecated, use 'type' for this " "API instead. See https://github.com/elastic/elasticsearch-py/" "issues/1698 for more information", category=DeprecationWarning, stacklevel=2, ) if doc_type_in_params: params["type"] = params.pop("doc_type") if doc_type_in_kwargs: kwargs["type"] = kwargs.pop("doc_type") if using_body_kwarg or using_positional_args: # If there are any body-only parameters then we raise a 'TypeError' # to alert the user they have to either not use a 'body' parameter # or to put the parameter into the body. body_only_params_in_use = body_only_params.intersection(kwargs) if body_only_params_in_use: # Make sure the error message prose makes sense! params_prose = "', '".join(sorted(body_only_params_in_use)) plural_params = len(body_only_params_in_use) > 1 raise TypeError( "The '%s' parameter%s %s only serialized in the request body " "and can't be combined with the 'body' parameter. Either stop using the " "'body' parameter and use keyword-arguments only or move the specified " "parameters into the 'body'. See https://github.com/elastic/elasticsearch-py/" "issues/1698 for more information" % ( params_prose, "s" if plural_params else "", "are" if plural_params else "is", ) ) # If there's no parameter overlap we still warn the user # that the 'body' parameter is deprecated for this API. if using_body_kwarg and body_params: warnings.warn( "The 'body' parameter is deprecated for the '%s' API and " "will be removed in a future version. Instead use API parameters directly. " "See https://github.com/elastic/elasticsearch-py/issues/1698 for " "more information" % str(func.__name__), DeprecationWarning, stacklevel=2, ) # If positional arguments are being used we also warn about that being deprecated. if using_positional_args: warnings.warn( "Using positional arguments for APIs is deprecated and will be " "disabled in 8.0.0. Instead use only keyword arguments for all APIs. " "See https://github.com/elastic/elasticsearch-py/issues/1698 for " "more information", DeprecationWarning, stacklevel=2, ) # We need to serialize all these parameters into a JSON body. elif set(body_params or ()).intersection(kwargs): body = {} for param in body_params: value = kwargs.pop(param, None) if value is not None: body[param.rstrip("_")] = value kwargs["body"] = body # Since we've deprecated 'body' we set body={} if there # should be a body on JSON-field APIs but none of those fields # are filled. elif body_required: kwargs["body"] = {} # If there's a named body parameter then we transform it to 'body' # for backwards compatibility with libraries like APM. # Otherwise we warn the user about 'body' being deprecated. if body_name: if body_name in kwargs: # If passed both 'body' and the named body param we raise an error. if using_body_kwarg: raise TypeError( "Can't use '%s' and 'body' parameters together because '%s' " "is an alias for 'body'. Instead you should only use the " "'%s' parameter. See https://github.com/elastic/elasticsearch-py/" "issues/1698 for more information" % ( body_name, body_name, body_name, ) ) kwargs["body"] = kwargs.pop(body_name) # Warn if user passes 'body' but should be using the named body parameter. elif using_body_kwarg: warnings.warn( "The 'body' parameter is deprecated for the '%s' API and " "will be removed in a future version. Instead use the '%s' parameter. " "See https://github.com/elastic/elasticsearch-py/issues/1698 " "for more information" % (str(func.__name__), body_name), category=DeprecationWarning, stacklevel=2, ) if http_auth is not None and api_key is not None: raise ValueError( "Only one of 'http_auth' and 'api_key' may be passed at a time" ) elif http_auth is not None: headers["authorization"] = "Basic %s" % ( _base64_auth_header(http_auth), ) elif api_key is not None: headers["authorization"] = "ApiKey %s" % (_base64_auth_header(api_key),) for p in es_query_params + GLOBAL_PARAMS: if p in kwargs: v = kwargs.pop(p) if v is not None: params[p] = _escape(v) # don't treat ignore, request_timeout, and opaque_id as other params to avoid escaping for p in ("ignore", "request_timeout"): if p in kwargs: params[p] = kwargs.pop(p) return func(*args, params=params, headers=headers, **kwargs) return _wrapped return _wrapper def _bulk_body(serializer, body): # if not passed in a string, serialize items and join by newline if not isinstance(body, string_types): body = "\n".join(map(serializer.dumps, body)) # bulk body must end with a newline if isinstance(body, bytes): if not body.endswith(b"\n"): body += b"\n" elif isinstance(body, string_types) and not body.endswith("\n"): body += "\n" return body def _base64_auth_header(auth_value): """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)): auth_value = base64.b64encode(to_bytes(":".join(auth_value))) return to_str(auth_value) class NamespacedClient(object): def __init__(self, client): self.client = client @property def transport(self): return self.client.transport class AddonClient(NamespacedClient): @classmethod def infect_client(cls, client): addon = cls(weakref.proxy(client)) setattr(client, cls.namespace, addon) return client