opensearch-py

To avoid loading massive datasets into memory as a single Python list, you can pass a generator function to helpers.bulk or helpers.parallel_bulk. This allows you to stream data into the bulk process efficiently. The generator should yield dictionaries containing the necessary metadata (like _index and _id) and the document body.

def _generate_data():
    for i in range(100):
        yield {"_index": index_name, "_id": i, "value": i}

succeeded = []
failed = []
for success, item in helpers.parallel_bulk(client, actions=_generate_data()):
    if success:
        succeeded.append(item)
    else:
        failed.append(item)

if len(failed) > 0:
    print(f"There were {len(failed)} errors:")
    for item in failed:
        print(item["index"]["error"])

if len(succeeded) > 0:
    print(f"Bulk-inserted {len(succeeded)} items (streaming_bulk).")

The OpenSearch Python client allows you to specify different connection classes depending on whether you are using the synchronous or asynchronous client, and which underlying HTTP library you prefer.

• Synchronous Client: Supports Urllib3HttpConnection (the default, based on urllib3) and RequestsHttpConnection (based on requests). Use Urllib3HttpConnection unless your application is already standardized on requests.
• Asynchronous Client: Uses AsyncHttpConnection (based on aiohttp) via the AsyncOpenSearch client.

In opensearch-py (version 2.2.0 and later), you can import high-level DSL functionalities directly from the opensearchpy.helpers module. This eliminates the requirement to import opensearch-dsl-py for the following feature sets:

• aggs
• analysis
• document
• faceted search
• field
• function
• index
• mapping
• query
• search
• update by query
• utils
• wrappers

To start using OpenSearch, you must first create a client instance using OpenSearch. You can then create indices using client.indices.create and populate them with documents using client.index. After adding documents, it is recommended to call client.indices.refresh to ensure the documents are immediately searchable.

from opensearchpy import OpenSearch

# create an OpenSearch client
client = OpenSearch(hosts=['localhost'])

# create an index
client.indices.create(index='movies')

# add a document
client.index(
    index='movies',
    body={
        'title': 'The Godfather',
        'director': 'Francis Ford Coppola',
        'year': 1972
    }
)

# refresh the index to make the documents searchable
client.indices.refresh(index='movies')

To perform snapshot actions, you first need to initialize an OpenSearch client and create an index to back up. This example demonstrates connecting to a local instance and creating a movies index.

from opensearchpy import OpenSearch

host = 'localhost'
port = 9200
auth = ('admin', 'admin') # For testing only. Don't store credentials in code.

client = OpenSearch(
    hosts = [{'host': host, 'port': port}],
    http_auth = auth,
    use_ssl = True,
    verify_certs = False,
    ssl_show_warn = False
)

print(client.info())  # Check server info and make sure the client is connected
client.indices.create(index='movies')

You can index vectors using the bulk API. Ensure that the vector being indexed matches the dimension defined in the index mapping.

vectors = []
for i in range(10):
    vec = []
    for j in range(dimensions): 
        vec.append(round(random.uniform(0, 1), 2)) 
  
    vectors.append({
        "_index": index_name,
        "_id": i,
        "values": vec,
    })

helpers.bulk(client, vectors)

client.indices.refresh(index=index_name)

To use Kerberos authentication, you can use third-party Python packages like requests-kerberos or requests-gssapi. Pass the appropriate authentication object to the http_auth parameter. Note that parameters like mutual_authentication may need to be adjusted based on your specific server configuration.

from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_kerberos import HTTPKerberosAuth, OPTIONAL

client = OpenSearch(
    ['https://...'],
    use_ssl=True,
    verify_certs=True,
    http_auth=HTTPKerberosAuth(mutual_authentication=OPTIONAL)
)

health = client.cluster.health()