in elastic_enterprise_search/_utils.py [0:0]
def wrapper(api: F) -> F:
@wraps(api)
def wrapped(*args: t.Any, **kwargs: t.Any) -> t.Any:
nonlocal api, body_name, body_fields
# 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 client 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
# 'http_auth' needs to be aliased to 'basic_auth' or 'bearer_auth'.
transport_option = option
if option == "http_auth":
if isinstance(kwargs["http_auth"], str):
transport_option = "bearer_auth"
elif (
isinstance(kwargs["http_auth"], (list, tuple))
and len(kwargs["http_auth"]) == 2
):
transport_option = "basic_auth"
else:
raise TypeError(
"'http_auth' must be either a str or a 2-tuple of strings"
)
try:
transport_options[transport_option] = kwargs.pop(option)
except KeyError:
pass
if transport_options:
client = args[0].options(**transport_options)
warnings.warn(
"Passing transport options in the API method is deprecated. "
f"Use '{type(client).__name__}.options()' instead.",
category=DeprecationWarning,
stacklevel=warn_stacklevel(),
)
args = (client,) + args[1:]
if "body" in kwargs and (
not ignore_deprecated_options or "body" not in ignore_deprecated_options
):
body = kwargs.pop("body")
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."
)
warnings.warn(
"The 'body' parameter is deprecated and will be removed "
f"in a future version. Instead use the '{body_name}' parameter.",
category=DeprecationWarning,
stacklevel=warn_stacklevel(),
)
kwargs[body_name] = body
elif body_fields:
if not hasattr(body, "items"):
raise ValueError(
"Couldn't merge 'body' with other parameters as it wasn't a mapping. "
"Instead of using 'body' use individual API parameters"
)
warnings.warn(
"The 'body' parameter is deprecated and will be removed "
"in a future version. Instead use individual parameters.",
category=DeprecationWarning,
stacklevel=warn_stacklevel(),
)
# Special handling of page:{current:1,size:20} -> current_page=1, page_size=20
if "page" in body and set(body["page"]).issubset(
{"current", "size"}
):
page = body.pop("page")
if "current" in page:
kwargs["current_page"] = page["current"]
if "size" in page:
kwargs["page_size"] = page["size"]
_merge_kwargs_no_duplicates(kwargs, body)
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]