Copy index using the Python/elasticsearch package

Hello, a quick post on how to copy an index using a python client in version 7 (this is also a demonstration of how to use clients in a lower version)

import argparse
from elasticsearch7 import Elasticsearch
from elasticsearch7.exceptions import ConnectionError
from datetime import datetime

def main():
    parser = argparse.ArgumentParser(description="Copy an Elasticsearch index with a new name and alias.")
    parser.add_argument("host", help="Elasticsearch host (e.g., http://localhost:9200)")
    parser.add_argument("source_index", help="Name of the source index to copy")

    args = parser.parse_args()

    # Elasticsearch host from command line argument
    host = args.host

    # Source index name from command line argument
    source_index = args.source_index


    try:
        # Create an Elasticsearch client
        client = Elasticsearch(host)
        
        # If no exception is raised, the client was successfully created
        print("Connection to Elasticsearch established.")

        # Get the current timestamp
        timestamp = datetime.now().strftime("%Y%m%d%H%M%S")

        # Retrieve the existing index details
        index_details = client.indices.get(index=source_index)
        
        # Get the provided_name from the index details
        provided_name = list(index_details.keys())[0]

        # New index name with a timestamp
        new_index_name = f"{provided_name}_{timestamp}"

        # Temporarily set the index.blocks.write setting to true
        client.indices.put_settings(index=provided_name, body={"index.blocks.write": True})

        # Clone the index with the provided_name
        client.indices.clone(index=provided_name, target=new_index_name)

        # Create a new alias for the original index
        new_alias_name = f"{source_index}_alias_{timestamp}"
        client.indices.put_alias(index=new_index_name, name=new_alias_name)

        # Reset the index.blocks.write setting to false
        client.indices.put_settings(index=provided_name, body={"index.blocks.write": False})

        print(f"Successfully created a copy of '{provided_name}' with the name '{new_index_name}' and alias '{new_alias_name}'.")

    except ConnectionError as e:
        print(f"Failed to establish a connection to Elasticsearch: {e}")


if __name__ == "__main__":
    main()
Show Comments
AbuseIPDB Contributor Badge LinkedIn Profile