Test Rest API through python

Elastic Stack Elasticsearch
21

Thanks, @navya_k!

You can try something like this?

from elasticsearch import Elasticsearch
from getpass import getpass
from datetime import datetime

# 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 based on a time range provided by the user.
def search_logs(index_name):
    # Get user input for start and end time
    start_time = input("Enter start time (YYYY-MM-DDTHH:MM:SS): ")
    end_time = input("Enter end time (YYYY-MM-DDTHH:MM:SS): ")

    # Convert the input times to ISO format
    start_time_iso = datetime.strptime(start_time, '%Y-%m-%dT%H:%M:%S').isoformat()
    end_time_iso = datetime.strptime(end_time, '%Y-%m-%dT%H:%M:%S').isoformat()

    # Construct the query
    query = {
        "range": {
            "timestamp": {
                "gte": start_time_iso,
                "lte": end_time_iso
            }
        }
    }

    # Execute the search query
    response = client.search(index=index_name, body={"query": query})
    return response['hits']['hits']

# 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 within a time range specified by the user
    search_results = search_logs(index_name)
    print(search_results)

Let me know if that works for you?

22

Lots of Hellos...
Thank you for reply.

I am getting output.

But when I use import json and print(json.dumps(results))...it's throwing error as API response is not JSON serializable.

How can we print only id/ service name/app_name/transactions list..

23

Thanks, @navya_k for all of your follow up here.

Maybe something like this might work for you:

from elasticsearch import Elasticsearch
from getpass import getpass
import json

def get_elasticsearch_client():
    return Elasticsearch(getpass("Host: "),
    api_key=getpass("API Key: "))

def search_logs(client, index_name, service_name, start_time, end_time):
    try:
        query = {
            "query": {
                "bool": {
                    "must": [
                        {"match": {"service_name": service_name}}
                    ],
                    "filter": [
                        {"range": {"timestamp": {"gte": start_time, "lte": end_time}}}
                    ]
                }
            },
            "_source": ["id", "service_name", "app_name", "transactions"]
        }
        response = client.search(index=index_name, body=query, size=100)
        return [hit['_source'] for hit in response['hits']['hits']]
    except Exception as e:
        print(f"Error searching logs: {e}")
        return []

if __name__ == "__main__":
    client = get_elasticsearch_client()
    index_name = "sample_logs"
    service_name = input("Enter the service name to filter logs: ")
    start_time = "2024-04-23T00:00:00"
    end_time = "2024-04-24T00:00:00"
    results = search_logs(client, index_name, service_name, start_time, end_time)
    print(json.dumps(results, indent=2))

My guess is that this script only processes an example dataset that lacks breadth. Taking a step back, I've found our Python Client Documentation to be very helpful. Here are a few other resources that you might find useful:

24

Thank you

1 Like