in elasticsearch/_sync/client/utils.py [0:0]
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]