Demo code for pushing documents

Prerequisites

Before you begin, make sure you have:

• An OpenSearch High-performance Search Edition application with at least one table

• Python 3 installed

• A RAM user with the required permissions. See AliyunServiceRoleForOpenSearch and Access authorization rules

Set up credentials

Store your AccessKey pair as environment variables. Never hardcode credentials in source code.

Linux and macOS

export ALIBABA_CLOUD_ACCESS_KEY_ID=<access_key_id>
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=<access_key_secret>

Replace <access_key_id> and <access_key_secret> with the AccessKey ID and AccessKey secret of your RAM user.

Windows

1. Create an environment variable file and add ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET with your AccessKey ID and AccessKey secret.

2. Restart Windows for the changes to take effect.

Install dependencies

pip install alibabacloud_tea_util
pip install alibabacloud_opensearch_util
pip install alibabacloud_credentials

The sample code also imports BaseRequest, which contains the Config and Client classes. BaseRequest is not a pip package — get the implementation from Sample code for the Python client.

Document operations

Each document in a bulk request requires a cmd field that specifies the operation:

Sample code

The following example initializes the client and pushes documents using all three operations.

# -*- coding: utf-8 -*-
import time, os
from typing import Dict, Any
from Tea.exceptions import TeaException
from Tea.request import TeaRequest
from alibabacloud_tea_util import models as util_models
from BaseRequest import Config, Client


class opensearch:
    def __init__(self, config: Config):
        self.Clients = Client(config=config)
        self.runtime = util_models.RuntimeOptions(
            connect_timeout=10000,
            read_timeout=10000,
            autoretry=False,
            ignore_ssl=False,
            max_idle_conns=50,
            max_attempts=3
        )
        self.header = {}

    def docBulk(self, app_name: str, table_name: str, doc_content: list) -> Dict[str, Any]:
        try:
            response = self.Clients._request(
                method="POST",
                pathname=f'/v3/openapi/apps/{app_name}/{table_name}/actions/bulk',
                query={},
                headers=self.header,
                body=doc_content,
                runtime=self.runtime
            )
            return response
        except Exception as e:
            print(e)


if __name__ == "__main__":
    # Endpoint of the OpenSearch API — do not include the http:// or https:// prefix
    endpoint = "<endpoint>"

    # Request protocol: HTTP or HTTPS
    endpoint_protocol = "HTTP"

    # Read credentials from environment variables
    access_key_id = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID")
    access_key_secret = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET")

    # Authentication type
    # access_key: authenticate with an AccessKey pair (default)
    # sts: authenticate with a Security Token Service (STS) token (required for RAM role assumption)
    auth_type = "sts"

    # STS token — required only when auth_type is "sts"
    # Obtain this token by calling the AssumeRole operation of Alibaba Cloud RAM
    security_token = "<security_token>"

    Configs = Config(
        endpoint=endpoint,
        access_key_id=access_key_id,
        access_key_secret=access_key_secret,
        security_token=security_token,  # Required only for STS authentication
        type=auth_type,
        protocol=endpoint_protocol
    )

    ops = opensearch(Configs)
    app_name = "app_name"
    table_name = "table_name"

    # --------------- Push documents ---------------

    # ADD: include timestamp (milliseconds) to control update order for documents
    # with the same primary key. Without timestamp, OpenSearch applies updates
    # in the order they arrive.
    document1 = {"cmd": "ADD", "timestamp": int(time.time() * 1000), "fields": {"id": "1", "title": "opensearch"}}
    document2 = {"cmd": "ADD", "fields": {"id": 2, "describe": "123456"}}

    # DELETE: only the primary key field is required
    deletedoc = {"cmd": "DELETE", "fields": {"id": 2}}

    # UPDATE: only the specified fields are updated; other fields are unchanged
    updatedoc = {"cmd": "UPDATE", "fields": {"id": 2, "describe": "6666", "title": "OpenSearch"}}

    documents = [document1, document2]
    res5 = ops.docBulk(app_name=app_name, table_name=table_name, doc_content=documents)
    print(res5)

What's next

For information about processing and transforming documents before pushing them, see Process data.