neptune/core/lib/opensearch.py

from django.conf import settings
from opensearchpy import OpenSearch


def initialise_opensearch():
    auth = (settings.OPENSEARCH_USERNAME, settings.OPENSEARCH_PASSWORD)
    client = OpenSearch(
        # fmt: off
        hosts=[{"host": settings.OPENSEARCH_URL,
                "port": settings.OPENSEARCH_PORT}],
        http_compress=False,  # enables gzip compression for request bodies
        http_auth=auth,
        # client_cert = client_cert_path,
        # client_key = client_key_path,
        use_ssl=settings.OPENSEARCH_TLS,
        verify_certs=False,
        ssl_assert_hostname=False,
        ssl_show_warn=False,
        # a_certs=ca_certs_path,
    )
    return client


def construct_query(query, fields, results):
    if not fields:
        fields = settings.OPENSEARCH_MAIN_SEARCH_FIELDS
    if not results:
        results = 5
    query = {
        "size": results,
        "query": {
            "multi_match": {
                "query": query,
                "fields": fields,
            }
        },
    }
    return query


def run_main_query(client, query, fields=None, results=None):
    search_query = construct_query(query, fields, results)
    # fmt: off
    response = client.search(body=search_query,
                             index=settings.OPENSEARCH_INDEX_MAIN)
    return response