Python client

출처: 문서

본문

Python 클라이언트

OpenSearch 저수준(low-level) Python 클라이언트(opensearch-py)는 OpenSearch REST API를 위한 래퍼 메서드를 제공해서 Python에서 클러스터와 더 자연스럽게 상호작용할 수 있게 해줘요. 주어진 URL로 원시(raw) HTTP 요청을 보내는 대신, 클러스터용 OpenSearch 클라이언트를 만들고 클라이언트의 내장 함수를 호출하면 돼요.

이 시작 가이드에서는 OpenSearch에 연결하고, 문서를 인덱싱하고, 쿼리를 실행하는 방법을 보여줘요. 추가 정보는 다음 리소스를 참고하세요:

• OpenSearch Python 저장소
• API 레퍼런스
• 사용자 가이드
• 샘플

질문이 있거나 기여하고 싶다면 이슈를 생성해서 OpenSearch Python 팀과 직접 소통할 수 있어요.

Python 클라이언트 설치하기

클라이언트의 최신 버전인 opensearch-py 3.2.0은 Python 3.10 이상이 필요해요. 프로젝트에 클라이언트를 추가하려면 pip로 설치하세요:

pip install opensearch-py

{% include copy.html %}

클라이언트를 설치한 후에는 다른 모듈처럼 임포트할 수 있어요:

from opensearchpy import OpenSearch

{% include copy.html %}

샘플 데이터

이 페이지의 예제에서는 학생 문서를 사용해요. 각 문서는 firstName, lastName, gpa, gradDate 필드를 담고 있는 Python 딕셔너리예요. 예를 들어 다음 딕셔너리는 한 명의 학생을 나타내요:

document = {'firstName': 'John', 'lastName': 'Doe', 'gpa': 3.89, 'gradDate': '2022-05-15'}

{% include copy.html %}

OpenSearch에 연결하기

기본 OpenSearch 호스트에 연결하려면, Security 플러그인을 사용하는 경우 SSL이 활성화된 클라이언트 객체를 만드세요. <custom-admin-password>는 OpenSearch를 설치할 때 설정한 관리자 비밀번호로 바꾸세요:

host = 'localhost'
port = 9200
auth = ('admin', '<custom-admin-password>') # For testing only. Don't store credentials in code.
ca_certs_path = '/full/path/to/root-ca.pem' # Provide a CA bundle if you use intermediate CAs with your root CA.

# Create the client with SSL/TLS enabled, but hostname verification disabled.
client = OpenSearch(
    hosts = [{'host': host, 'port': port}],
    http_compress = True, # enables gzip compression for request bodies
    http_auth = auth,
    use_ssl = True,
    verify_certs = True,
    ssl_assert_hostname = False,
    ssl_show_warn = False,
    ca_certs = ca_certs_path
)

{% include copy.html %}

자체 클라이언트 인증서가 있다면 client_cert_path와 client_key_path 파라미터에 지정하세요:

host = 'localhost'
port = 9200
auth = ('admin', '<custom-admin-password>') # For testing only. Don't store credentials in code.
ca_certs_path = '/full/path/to/root-ca.pem' # Provide a CA bundle if you use intermediate CAs with your root CA.

# Optional client certificates if you don't want to use HTTP basic authentication.
client_cert_path = '/full/path/to/client.pem'
client_key_path = '/full/path/to/client-key.pem'

# Create the client with SSL/TLS enabled, but hostname verification disabled.
client = OpenSearch(
    hosts = [{'host': host, 'port': port}],
    http_compress = True, # enables gzip compression for request bodies
    http_auth = auth,
    client_cert = client_cert_path,
    client_key = client_key_path,
    use_ssl = True,
    verify_certs = True,
    ssl_assert_hostname = False,
    ssl_show_warn = False,
    ca_certs = ca_certs_path
)

{% include copy.html %}

Security 플러그인을 사용하지 않는다면 SSL이 비활성화된 클라이언트 객체를 만드세요:

host = 'localhost'
port = 9200

# Create the client with SSL/TLS and hostname verification disabled.
client = OpenSearch(
    hosts = [{'host': host, 'port': port}],
    http_compress = True, # enables gzip compression for request bodies
    use_ssl = False,
    verify_certs = False,
    ssl_assert_hostname = False,
    ssl_show_warn = False
)

{% include copy.html %}

Amazon OpenSearch Service에 연결하기

IAM 자격 증명을 사용해 Amazon OpenSearch Service나 Amazon OpenSearch Serverless로 가는 요청에 서명하려면, Python용 AWS SDK(Boto3)를 설치하세요:

pip install boto3

{% include copy.html %}

다음 예제에서는 엔드포인트를 여러분의 도메인 엔드포인트로 바꾸세요. 도메인 엔드포인트는 Amazon OpenSearch Service 콘솔의 도메인 세부 정보 페이지에 표시돼요.

다음 예제는 IAM 자격 증명을 사용해 Amazon OpenSearch Service에 연결하는 방법을 보여줘요:

from opensearchpy import OpenSearch, RequestsHttpConnection, RequestsAWSV4SignerAuth
import boto3

host = 'search-<domain-name>-<id>.us-east-1.es.amazonaws.com' # Domain endpoint without https://
region = 'us-east-1'
service = 'es'
credentials = boto3.Session().get_credentials()
auth = RequestsAWSV4SignerAuth(credentials, region, service)

client = OpenSearch(
    hosts = [{'host': host, 'port': 443}],
    http_auth = auth,
    use_ssl = True,
    verify_certs = True,
    connection_class = RequestsHttpConnection,
    pool_maxsize = 20
)

{% include copy.html %}

사용자 이름과 비밀번호로 HTTP를 통해 Amazon OpenSearch Service에 연결하려면 다음 코드를 사용하세요:

from opensearchpy import OpenSearch

host = 'search-<domain-name>-<id>.us-east-1.es.amazonaws.com' # Domain endpoint without https://
auth = ('admin', '<custom-admin-password>') # For testing only. Don't store credentials in code.

client = OpenSearch(
    hosts=[{"host": host, "port": 443}],
    http_auth=auth,
    http_compress=True,  # enables gzip compression for request bodies
    use_ssl=True,
    verify_certs=True,
    ssl_assert_hostname=False,
    ssl_show_warn=False,
)

{% include copy.html %}

Amazon OpenSearch Serverless에 연결하기

다음 예제에서는 엔드포인트를 여러분의 컬렉션 엔드포인트로 바꾸세요. 컬렉션 엔드포인트는 Amazon OpenSearch Service 콘솔의 컬렉션 세부 정보 페이지에 표시돼요.

다음 예제는 Amazon OpenSearch Serverless에 연결하는 방법을 보여줘요:

from opensearchpy import OpenSearch, RequestsHttpConnection, RequestsAWSV4SignerAuth
import boto3

host = '<collection-id>.us-east-1.aoss.amazonaws.com' # Collection endpoint without https://
region = 'us-east-1'
service = 'aoss'
credentials = boto3.Session().get_credentials()
auth = RequestsAWSV4SignerAuth(credentials, region, service)

client = OpenSearch(
    hosts = [{'host': host, 'port': 443}],
    http_auth = auth,
    use_ssl = True,
    verify_certs = True,
    connection_class = RequestsHttpConnection,
    pool_maxsize = 20
)

{% include copy.html %}

Amazon OpenSearch Serverless는 OpenSearch API 작업의 일부만 지원하며, 이 페이지의 예제에서 사용하는 refresh 파라미터는 지원하지 않아요. 자세한 내용은 Amazon OpenSearch Serverless에서 지원되는 작업 및 플러그인을 참고하세요. {: .note}

인덱스 생성하기

다음 예제는 프라이머리 샤드 하나와 레플리카 하나로 인덱스를 만들어요. gradDate 필드를 yyyy-MM-dd 형식의 date로 명시적으로 매핑해요. 문서를 인덱싱할 때 나머지 문서 필드는 OpenSearch가 동적으로 매핑해요:

index_name = 'students'
index_body = {
    'settings': {
        'index': {
            'number_of_shards': 1,
            'number_of_replicas': 1
        }
    },
    'mappings': {
        'properties': {
            'gradDate': {'type': 'date', 'format': 'yyyy-MM-dd'}
        }
    }
}
response = client.indices.create(index=index_name, body=index_body)

{% include copy.html %}

문서 인덱싱하기

샘플 데이터의 document 딕셔너리를 인덱싱하려면 client.index() 메서드를 사용하세요:

response = client.index(index=index_name, id='1', body=document, refresh=True)

{% include copy.html %}

벌크 작업 수행하기

클라이언트의 bulk() 메서드를 사용하면 여러 작업을 한 번에 수행할 수 있어요. 작업들은 같은 유형일 수도 있고 서로 다른 유형일 수도 있어요. 각 액션 뒤에 해당 문서가 오는 리스트 형태로 작업을 전달하세요:

operations = [
    {'index': {'_index': index_name, '_id': '2'}},
    {'firstName': 'Paulo', 'lastName': 'Santos', 'gpa': 3.93, 'gradDate': '2021-05-20'},
    {'index': {'_index': index_name, '_id': '3'}},
    {'firstName': 'Shirley', 'lastName': 'Rodriguez', 'gpa': 3.91, 'gradDate': '2019-05-10'}
]
response = client.bulk(body=operations, refresh=True)

{% include copy.html %}

문서 검색하기

인덱스의 모든 문서를 검색하려면 쿼리 없이 client.search() 메서드를 사용하세요:

response = client.search(index=index_name)

{% include copy.html %}

응답은 딕셔너리이고, response['hits']['hits']의 각 항목은 _id 키에 문서 ID를, _source 키에 문서 필드를 담고 있는 딕셔너리예요:

for hit in response['hits']['hits']:
    source = hit['_source']
    print(f"ID: {hit['_id']}, name: {source['firstName']} {source['lastName']}, GPA: {source['gpa']}, graduation date: {source['gradDate']}")

{% include copy.html %}

쿼리로 검색하려면 요청 본문에 쿼리를 담아 전달하세요. 다음 코드는 range 쿼리를 사용해 2019년에 졸업한 학생을 검색해요:

query = {'query': {'range': {'gradDate': {'gte': '2019-01-01', 'lte': '2019-12-31'}}}}
response = client.search(index=index_name, body=query)

{% include copy.html %}

결과 페이징하기

결과를 페이징하려면 from과 size 파라미터를 사용하세요. 다음 예제는 학생을 졸업 날짜 기준으로 정렬하고 한 번에 두 개씩 결과를 가져와요. 첫 번째 요청은 결과의 첫 페이지를 반환하고, 두 번째 요청은 다음 페이지를 반환해요:

response = client.search(index=index_name, body={'from': 0, 'size': 2, 'sort': [{'gradDate': 'asc'}]})
next_response = client.search(index=index_name, body={'from': 2, 'size': 2, 'sort': [{'gradDate': 'asc'}]})
for page in [response, next_response]:
    for hit in page['hits']['hits']:
        print(hit['_source'])

{% include copy.html %}

from과 size 파라미터는 결과의 첫 몇 페이지를 다룰 때 잘 동작해요. 많은 수의 결과를 페이징하려면 point in time과 search_after를 사용하세요. 자세한 내용은 결과 페이징을 참고하세요.

문서 업데이트하기

client.update() 메서드로 문서를 업데이트할 수 있어요. doc 객체의 필드는 기존 문서에 병합돼요:

response = client.update(index=index_name, id='1', body={'doc': {'gpa': 3.92}})

{% include copy.html %}

문서 삭제하기

client.delete() 메서드로 문서를 삭제할 수 있어요:

response = client.delete(index=index_name, id='3', refresh=True)

{% include copy.html %}

인덱스 삭제하기

client.indices.delete() 메서드로 인덱스를 삭제할 수 있어요:

response = client.indices.delete(index=index_name)

{% include copy.html %}

샘플 프로그램

이 샘플 프로그램은 앞선 섹션들의 코드를 합친 거예요. Security 플러그인이 활성화된 클러스터에 연결해요. Security 플러그인 없는 클러스터에 연결하려면 # Without security 주석이 달린 줄을 바꾸세요.

이 샘플 프로그램은 테스트용이에요. 코드에 자격 증명을 직접 지정하고 있어요. 프로덕션 환경에서는 안전한 위치에서 자격 증명을 불러오세요. {: .warning}

다음 샘플 프로그램은 클라이언트를 만들고, 인덱스를 생성하고, 문서를 개별적으로 그리고 벌크로 인덱싱하고, 문서를 검색하고, 문서를 업데이트하고, 문서를 삭제한 다음 인덱스를 삭제해요:

import json
from opensearchpy import OpenSearch

host = 'localhost'
port = 9200
# Without security, remove this line
auth = ('admin', '<custom-admin-password>')
# Without security, remove this line
ca_certs_path = '/full/path/to/root-ca.pem' # Provide a CA bundle if you use intermediate CAs with your root CA.

# Create the client with SSL/TLS enabled, but hostname verification disabled.
client = OpenSearch(
    hosts = [{'host': host, 'port': port}],
    http_compress = True, # enables gzip compression for request bodies
    http_auth = auth, # Without security, remove this line
    use_ssl = True, # Without security, use use_ssl = False
    verify_certs = True, # Without security, use verify_certs = False
    ssl_assert_hostname = False,
    ssl_show_warn = False,
    ca_certs = ca_certs_path # Without security, remove this line
)

# Create the index.
index_name = 'students'
index_body = {
    'settings': {
        'index': {
            'number_of_shards': 1,
            'number_of_replicas': 1
        }
    },
    'mappings': {
        'properties': {
            'gradDate': {'type': 'date', 'format': 'yyyy-MM-dd'}
        }
    }
}
print('Creating index......')
response = client.indices.create(index=index_name, body=index_body)
print(f"Index created: {response['index']}")

# Index a document.
print('\nIndexing one student......')
document = {'firstName': 'John', 'lastName': 'Doe', 'gpa': 3.89, 'gradDate': '2022-05-15'}
response = client.index(index=index_name, id='1', body=document, refresh=True)
print(f"Result: {response['result']}, id: {response['_id']}, version: {response['_version']}")

# Bulk index documents.
print('\nIndexing many students......')
operations = [
    {'index': {'_index': index_name, '_id': '2'}},
    {'firstName': 'Paulo', 'lastName': 'Santos', 'gpa': 3.93, 'gradDate': '2021-05-20'},
    {'index': {'_index': index_name, '_id': '3'}},
    {'firstName': 'Shirley', 'lastName': 'Rodriguez', 'gpa': 3.91, 'gradDate': '2019-05-10'}
]
response = client.bulk(body=operations, refresh=True)
print(f"Errors: {str(response['errors']).lower()}")
for item in response['items']:
    print(f"  {item['index']['result']} id: {item['index']['_id']}")

# Search for all students.
print('\nSearching for all students......')
for page, start in enumerate([0, 2], start=1):
    response = client.search(index=index_name, body={'from': start, 'size': 2, 'sort': [{'gradDate': 'asc'}]})
    if page == 1:
        print(f"Total hits: {response['hits']['total']['value']}")
    print(f"Page {page}:")
    for hit in response['hits']['hits']:
        print(f"  {json.dumps(hit['_source'], separators=(',', ':'))}")

# Search for students who graduated in 2019.
print('\nSearching for students who graduated in 2019......')
query = {'query': {'range': {'gradDate': {'gte': '2019-01-01', 'lte': '2019-12-31'}}}}
response = client.search(index=index_name, body=query)
print(f"Total hits: {response['hits']['total']['value']}")
for hit in response['hits']['hits']:
    print(f"  {json.dumps(hit['_source'], separators=(',', ':'))}")

# Update a document.
print("\nUpdating a student's GPA......")
response = client.update(index=index_name, id='1', body={'doc': {'gpa': 3.92}})
print(f"Result: {response['result']}, version: {response['_version']}")

# Get the updated document.
response = client.get(index=index_name, id='1')
print(f"Updated document: {json.dumps(response['_source'], separators=(',', ':'))}")

# Delete a document.
print('\nDeleting a student......')
response = client.delete(index=index_name, id='3', refresh=True)
print(f"Result: {response['result']}")

# Delete the index.
print('\nDeleting the index......')
response = client.indices.delete(index=index_name)
print(f"Acknowledged: {str(response['acknowledged']).lower()}")

{% include copy.html %}

프로그램은 다음과 같은 출력을 생성해요:

Creating index......
Index created: students

Indexing one student......
Result: created, id: 1, version: 1

Indexing many students......
Errors: false
  created id: 2
  created id: 3

Searching for all students......
Total hits: 3
Page 1:
  {"firstName":"Shirley","lastName":"Rodriguez","gpa":3.91,"gradDate":"2019-05-10"}
  {"firstName":"Paulo","lastName":"Santos","gpa":3.93,"gradDate":"2021-05-20"}
Page 2:
  {"firstName":"John","lastName":"Doe","gpa":3.89,"gradDate":"2022-05-15"}

Searching for students who graduated in 2019......
Total hits: 1
  {"firstName":"Shirley","lastName":"Rodriguez","gpa":3.91,"gradDate":"2019-05-10"}

Updating a student's GPA......
Result: updated, version: 2
Updated document: {"firstName":"John","lastName":"Doe","gpa":3.92,"gradDate":"2022-05-15"}

Deleting a student......
Result: deleted

Deleting the index......
Acknowledged: true

관련 문서

• Python에서 데이터를 분석하고 ML 모델을 업로드하려면 Python ML 클라이언트를 참고하세요.
• 클라이언트 API 레퍼런스는 opensearch-py API 문서를 참고하세요.
• 클라이언트 사용 예제가 더 필요하면 opensearch-py 사용자 가이드를 참고하세요.
• 벌크 인덱싱, 검색 같은 특정 작업의 가이드는 opensearch-py 가이드를 참고하세요.
• 완전한 샘플 애플리케이션은 opensearch-py 샘플을 참고하세요.

더 알아보기 (Learn more)

• 원문 전체: https://docs.opensearch.org/latest/clients/python-low-level/