# SPDX-License-Identifier: Apache-2.0
#
# The OpenSearch Contributors require contributions made to
# this file be licensed under the Apache-2.0 license or a
# compatible open source license.
#
# Modifications Copyright OpenSearch Contributors. See
# GitHub history for details.
from typing import Any
from ..client.utils import SKIP_IN_PATH, NamespacedClient, _make_path, query_params
[docs]class IndexManagementClient(NamespacedClient):
[docs] @query_params()
def put_policy(
self, policy: Any, body: Any = None, params: Any = None, headers: Any = None
) -> Any:
"""
Creates, or updates, a policy.
:arg policy: The name of the policy
"""
if policy in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'policy'.")
return self.transport.perform_request(
"PUT",
_make_path("_plugins", "_ism", "policies", policy),
params=params,
headers=headers,
body=body,
)
[docs] @query_params()
def add_policy(
self, index: Any, body: Any = None, params: Any = None, headers: Any = None
) -> Any:
"""
Adds a policy to an index. This operation does not change the policy if the index already has one.
:arg index: The name of the index to add policy on
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request(
"POST",
_make_path("_plugins", "_ism", "add", index),
params=params,
headers=headers,
body=body,
)
[docs] @query_params()
def get_policy(
self, policy: Any = None, params: Any = None, headers: Any = None
) -> Any:
"""
Gets the policy by `policy_id`; returns all policies if no policy_id is provided.
:arg policy: The name of the policy
"""
return self.transport.perform_request(
"GET",
_make_path("_plugins", "_ism", "policies", policy),
params=params,
headers=headers,
)
[docs] @query_params()
def remove_policy_from_index(
self, index: Any, params: Any = None, headers: Any = None
) -> Any:
"""
Removes any ISM policy from the index.
:arg index: The name of the index to remove policy on
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request(
"POST",
_make_path("_plugins", "_ism", "remove", index),
params=params,
headers=headers,
)
[docs] @query_params()
def change_policy(
self, index: Any, body: Any = None, params: Any = None, headers: Any = None
) -> Any:
"""
Updates the managed index policy to a new policy (or to a new version of the policy).
:arg index: The name of the index to change policy on
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request(
"POST",
_make_path("_plugins", "_ism", "change_policy", index),
params=params,
headers=headers,
body=body,
)
[docs] @query_params()
def retry(
self, index: Any, body: Any = None, params: Any = None, headers: Any = None
) -> Any:
"""
Retries the failed action for an index.
:arg index: The name of the index whose is in a failed state
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request(
"POST",
_make_path("_plugins", "_ism", "retry", index),
params=params,
headers=headers,
body=body,
)
[docs] @query_params("show_policy")
def explain_index(self, index: Any, params: Any = None, headers: Any = None) -> Any:
"""
Gets the current state of the index.
:arg index: The name of the index to explain
"""
if index in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'index'.")
return self.transport.perform_request(
"GET",
_make_path("_plugins", "_ism", "explain", index),
params=params,
headers=headers,
)
[docs] @query_params()
def delete_policy(
self, policy: Any, params: Any = None, headers: Any = None
) -> Any:
"""
Deletes the policy by `policy_id`.
:arg policy: The name of the policy to delete
"""
if policy in SKIP_IN_PATH:
raise ValueError("Empty value passed for a required argument 'policy'.")
return self.transport.perform_request(
"DELETE",
_make_path("_plugins", "_ism", "policies", policy),
params=params,
headers=headers,
)