Xiangping Liu

import json
import os
#from redb.settings import elasticsearch as es
from elasticsearch import Elasticsearch
from dataclasses import asdict
from dotenv import load_dotenv

class Utils():
    """
    Utility class for pushing

    """
    @staticmethod
    def push_to_elastic(sample_dict):
        """
        Pushes the sample_dict to elastic search
        """
        load_dotenv()
        ELASTIC_ENDPOINT = os.getenv("ELASTIC_HOST")
        ELASTIC_ENFORCE_SSL = os.getenv("ELASTIC_ENFORCE_SSL", False) == "True"
        ELASTIC_USER = os.getenv("ELASTIC_USER")
        ELASTIC_PASSWORD = os.getenv("ELASTIC_PASSWORD")
        ELASTIC_BINARIES_COLLECTION = os.getenv(
            "ELASTIC_BINARIES_COLLECTION", "redb_test"
        )
        es = Elasticsearch(
            ELASTIC_ENDPOINT,
            verify_certs=ELASTIC_ENFORCE_SSL,
            basic_auth=(ELASTIC_USER, ELASTIC_PASSWORD),
            maxsize=20,
            retry_on_timeout=True,
            timeout=30,
        )

        try:
            dict_dump = json.dumps(sample_dict)
        except Exception as e:
            print(f"Error: {e}")
            return False
        elastic = es.index(index=ELASTIC_BINARIES_COLLECTION, body=dict_dump)
        return True
    
    def push_to_index(sample_dict, index_name):
        """
        Pushes the sample_dict to elastic search
        """
        load_dotenv()
        ELASTIC_ENDPOINT = os.getenv("ELASTIC_HOST")
        ELASTIC_ENFORCE_SSL = os.getenv("ELASTIC_ENFORCE_SSL", False) == "True"
        ELASTIC_USER = os.getenv("ELASTIC_USER")
        ELASTIC_PASSWORD = os.getenv("ELASTIC_PASSWORD")
        ELASTIC_BINARIES_COLLECTION = index_name
        
        es = Elasticsearch(
            ELASTIC_ENDPOINT,
            verify_certs=ELASTIC_ENFORCE_SSL,
            basic_auth=(ELASTIC_USER, ELASTIC_PASSWORD),
            maxsize=20,
            retry_on_timeout=True,
            timeout=30,
        )

        try:
            dict_dump = json.dumps(sample_dict)
        except Exception as e:
            print(f"Error: {e}")
            return False
        elastic = es.index(index=ELASTIC_BINARIES_COLLECTION, body=dict_dump)
        return True