Test Rest API through python

Elastic Stack Elasticsearch
14

Thanks for following up here, @navya_k. As @dadoonet mentioned, it's easier to keep everything in one thread. I and others from Elastic usually block out some time daily to answer questions, but sometimes, there is some back-and-forth to find a solution.

I had a chance to look at the code you provided and was wondering if something like this would work:

from elasticsearch import Elasticsearch

# Initialize the Elasticsearch client
es = Elasticsearch("https://localhost:9200", api_key="example_api_key")

# Define the index pattern
index_pattern = "test-*"

# Input for the service name
service_name = input("Enter service name: ")

# Define the search query
query = {
    "query": {
        "bool": {
            "must": [
                {
                    "range": {
                        "@timestamp": {
                            "gte": "now-1h"
                        }
                    }
                },
                {
                    "term": {
                        "service.name": service_name
                    }
                }
            ]
        }
    }
}

# Execute the search query
result = es.search(index=index_pattern, body=query)

# Print each hit's source data
print("Search Results:")
for hit in result['hits']['hits']:  # Correct the path to hits
    print(hit['_source'])

I also had a few follow up questions:

  • What is the role of service_name in your query?
  • What errors are you getting, if any?
  • How are you hosting Elasticsearch?

I also wrote a simple script that takes in some sample logs and uploads them to Elasticsearch:

from elasticsearch import Elasticsearch
from getpass import getpass

# Initialize Elasticsearch client
client = Elasticsearch(
    getpass("Host: "),
    api_key=getpass("Elastic API Key: "),
)

# Uploads logs to a specified Elasticsearch index.
def upload_logs(index_name, logs):
    for log in logs:
        client.index(index=index_name, document=log)

#  Searches logs in a specified Elasticsearch index.
def search_logs(index_name, query):
    return client.search(index=index_name, query={"match": query})

# Example usage:
if __name__ == "__main__":
    # Define the index name
    index_name = "sample_logs"

    # Example log entries
    logs = [
        {"timestamp": "2024-04-23T12:00:00", "level": "INFO", "message": "System start."},
        {"timestamp": "2024-04-23T12:05:00", "level": "ERROR", "message": "Failed to connect to database."},
    ]

    # Upload logs
    upload_logs(index_name, logs)

    # Search logs
    search_results = search_logs(index_name, {"message": "database"})
    print(search_results)

Thanks again! My coworker @iulia and I also have a repository of Python scripts we use regularly, which could be a helpful resource here.