async.md

• Asynchronous I/O
  • Setup
  • Async Loop
  • Connect to OpenSearch
  • Create an Index
  • Index Documents
  • Refresh the Index
  • Search
  • Delete Documents
  • Delete the Index

Asynchronous I/O

This client supports asynchronous I/O that improves performance and increases throughput. See hello_async.py or knn_async_basics.py for a working asynchronous sample.

Setup

To add the async client to your project, install it using pip:

pip install opensearch-py[async]

In general, we recommend using a package manager, such as poetry, for your projects. This is the package manager used for samples. The following example includes opensearch-py[async] in pyproject.toml.

[tool.poetry.dependencies]
opensearch-py = { path = "../", extras=["async"] }

Async Loop

import asyncio

async def main():
    client = AsyncOpenSearch(...)
    try:
        # your code here
    finally:
        client.close()
    
if __name__ == "__main__":
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(main())
    loop.close()

Connect to OpenSearch

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

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

info = await client.info()
print(f"Welcome to {info['version']['distribution']} {info['version']['number']}!")

Create an Index

index_name = 'test-index'

index_body = {
    'settings': {
        'index': {
            'number_of_shards': 4
        }
    }
}

if not await client.indices.exists(index=index_name):
    await client.indices.create(
        index_name, 
        body=index_body
    )

Index Documents

await asyncio.gather(*[
    client.index(
        index = index_name,
        body = {
            'title': f"Moneyball {i}",
            'director': 'Bennett Miller',
            'year': '2011'
        },
        id = i
    ) for i in range(10)
])

Refresh the Index

await client.indices.refresh(index=index_name)

Search

    q = 'miller'

    query = {
        'size': 5,
        'query': {
            'multi_match': {
                'query': q,
                'fields': ['title^2', 'director']
            }
        }
    }

    results = await client.search(
        body = query,
        index = index_name
    )

    for hit in results["hits"]["hits"]:
      print(hit)

Delete Documents

await asyncio.gather(*[
    client.delete(
        index = index_name,
        id = i
    ) for i in range(10)
])

Delete the Index

await client.indices.delete(
    index = index_name
)