Demo code for querying hints

Prerequisites

Before you begin, make sure you have:

• An OpenSearch application (you'll need its name as app_name)

• The endpoint URL for your OpenSearch instance

• An Alibaba Cloud AccessKey pair — either from your Alibaba Cloud account or from a Resource Access Management (RAM) user with the required permissions

  • To create an AccessKey pair, see Create an AccessKey pair

  • To create a RAM user, see Create a RAM user

  • To grant the RAM user the required permissions, see AliyunServiceRoleForOpenSearch and Access authorization rules

Set environment variables

Linux and macOS

Replace <access_key_id> and <access_key_secret> with your AccessKey ID and AccessKey secret.

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

Windows

1. Create an environment variable file.

2. Add ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET to the file, and set them to your AccessKey ID and AccessKey secret.

3. Restart Windows for the changes to take effect.

Install dependencies

pip install alibabacloud_tea_util
pip install alibabacloud_opensearch_util
pip install alibabacloud_credentials

Sample code

The following code queries hints using the /v3/openapi/apps/{app_name}/actions/hint endpoint. Change app_name, endpoint, and the query parameters to match your application.

# -*- coding: utf-8 -*-
import 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 appAlgoSearch(self, app_name: str, algo_name: str, algo_type: str, query_params: dict) -> Dict[str, Any]:
        # Default algo_type is "suggest". For hints and top searches, use "hint" or "hot".
        algo_type = algo_type if algo_type else "suggest"
        path_name = f'/v3/openapi/apps/{app_name}/suggest/{algo_name}/search'
        if algo_type != "suggest":
            if algo_type not in ["hint", "hot"]:
                raise Exception("algo_type should be hint or hot")
            path_name = f'/v3/openapi/apps/{app_name}/actions/{algo_type}'
        try:
            response = self.Clients._request(
                method="GET",
                pathname=path_name,
                query=query_params,
                headers=self.header,
                body=None,
                runtime=self.runtime
            )
            return response
        except TeaException as e:
            print(e)


if __name__ == "__main__":
    # The endpoint does not include the http:// or https:// prefix.
    endpoint = "<endpoint>"

    # Valid values: HTTPS, HTTP
    endpoint_protocol = "HTTP"

    # Read credentials from environment variables set in the previous step.
    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 directly with an AccessKey pair (default).
    # "sts": authenticate with a Security Token Service (STS) token obtained via AssumeRole.
    auth_type = "access_key"

    # Required only when auth_type is "sts".
    security_token = "<security_token>"

    config = Config(
        endpoint=endpoint,
        access_key_id=access_key_id,
        access_key_secret=access_key_secret,
        security_token=security_token,
        type=auth_type,
        protocol=endpoint_protocol
    )

    ops = opensearch(config)
    app_name = "<app_name>"

    # Query hints
    hint_query = {
        "sort_type": "default",
        "hit": 10,
        "user_id": "a7a0d37c824b659f36a5b9e3b819fcdd"
    }
    result = ops.appAlgoSearch(
        app_name=app_name,
        algo_name=None,
        algo_type="hint",
        query_params=hint_query
    )
    print(result)

What's next

• Query top searches and hints — full API reference for the hints and top searches endpoints.