Python 개발 환경 구성

이 포스트에서는 OpenSearch RAG 실습을 위한 Python 프로젝트 구조, 패키지 설치, 환경 변수 관리 방식을 정의하고 클라이언트를 연결해 보는 과정을 다룹니다.

실습 기준은 Python 3.12 이상입니다.

권장 프로젝트 구조 #

opensearch-rag-lab/
  .env
  .gitignore
  pyproject.toml
  README.md
  app/
    __init__.py
    settings.py
    client.py
    healthcheck.py
    index_docs.py
    search_docs.py

opensearch-rag-lab/
  .env
  .gitignore
  pyproject.toml
  README.md
  app/
    __init__.py
    settings.py
    client.py
    healthcheck.py
    index_docs.py
    search_docs.py

가상환경 생성 #

python -m venv .venv
source .venv/bin/activate

python -m venv .venv
source .venv/bin/activate

uv 사용 가능하다면 uv로 진행

패키지 설치 #

pip install opensearch-py python-dotenv

pip install opensearch-py python-dotenv

패키지 설치 후 아래 순서로 OpenSearch 구동 테스트를 진행합니다.

.env 파일에 접속 정보를 적어 둡니다.
settings.py가 .env 값을 읽어서 Python 객체로 정리합니다.
client.py가 설정값을 사용해서 OpenSearch 클라이언트를 만듭니다.
healthcheck.py가 클라이언트를 사용해 OpenSearch 서버에 실제로 접속해 봅니다.
접속에 성공하면 클러스터 이름, 버전, 상태값이 출력됩니다.

설정 #

.env

OPENSEARCH_HOST=localhost
OPENSEARCH_PORT=9200
OPENSEARCH_USERNAME=admin
OPENSEARCH_PASSWORD=ChangeMe_12345!
OPENSEARCH_USE_SSL=TRUE
OPENSEARCH_VERIFY_CERTS=FALSE
OPENSEARCH_INDEX_NAME=rag_docs_v1

OPENSEARCH_HOST=localhost
OPENSEARCH_PORT=9200
OPENSEARCH_USERNAME=admin
OPENSEARCH_PASSWORD=ChangeMe_12345!
OPENSEARCH_USE_SSL=TRUE
OPENSEARCH_VERIFY_CERTS=FALSE
OPENSEARCH_INDEX_NAME=rag_docs_v1

settings.py

import os
from dataclasses import dataclass
from dotenv import load_dotenv

load_dotenv()

def _get_bool(name: str, default: bool = False) -> bool:
    value = os.getenv(name)
    if value is None:
        return default

    return value.strip().lower() in {"1", "true", "yes", "y", "on"}

@dataclass(frozen=True)
class OpenSearchSettings:
    host: str = os.getenv("OPENSEARCH_HOST", "localhost")
    port: int = int(os.getenv("OPENSEARCH_PORT", "9200"))
    username: str = os.getenv("OPENSEARCH_USERNAME", "admin")
    password: str = os.getenv(
        "OPENSEARCH_PASSWORD", os.getenv("OPENSEARCH_INITIAL_ADMIN_PASSWORD", "")
    )
    use_ssl: bool = _get_bool("OPENSEARCH_USE_SSL", True)
    verify_certs: bool = _get_bool("OPENSEARCH_VERIFY_CERTS", False)
    index_name: str = os.getenv("OPENSEARCH_INDEX_NAME", "rag_docs_v1")

import os
from dataclasses import dataclass
from dotenv import load_dotenv

load_dotenv()

def _get_bool(name: str, default: bool = False) -> bool:
    value = os.getenv(name)
    if value is None:
        return default

    return value.strip().lower() in {"1", "true", "yes", "y", "on"}

@dataclass(frozen=True)
class OpenSearchSettings:
    host: str = os.getenv("OPENSEARCH_HOST", "localhost")
    port: int = int(os.getenv("OPENSEARCH_PORT", "9200"))
    username: str = os.getenv("OPENSEARCH_USERNAME", "admin")
    password: str = os.getenv(
        "OPENSEARCH_PASSWORD", os.getenv("OPENSEARCH_INITIAL_ADMIN_PASSWORD", "")
    )
    use_ssl: bool = _get_bool("OPENSEARCH_USE_SSL", True)
    verify_certs: bool = _get_bool("OPENSEARCH_VERIFY_CERTS", False)
    index_name: str = os.getenv("OPENSEARCH_INDEX_NAME", "rag_docs_v1")

settings.py는 환경설정을 한 곳에서 관리하는 파일입니다.

load_dotenv()는 .env 파일에 적힌 값을 현재 Python 프로그램에서 읽을 수 있게 해줍니다.
_get_bool() 함수는 TRUE, FALSE처럼 문자열로 들어온 환경 변수 값을 Python의 True, False 값으로 바꿔 줍니다.
OpenSearchSettings 클래스는 host, port, username, password, index name 같은 설정값을 하나의 객체로 묶어 줍니다.
@dataclass(frozen=True)는 설정 객체를 만든 뒤 실수로 값을 바꾸지 못하게 하는 역할을 합니다.

쉽게 말해, settings.py는 코드 곳곳에 접속 정보를 흩뿌리지 않고 한 곳에서 안전하게 읽어 오기 위한 파일입니다.

설정 관리 원칙 #

비밀번호, API key, endpoint는 코드에 직접 쓰지 않습니다.
로컬 개발은 .env를 사용합니다.
운영 환경은 secret manager 또는 배포 플랫폼의 secret 기능을 사용합니다.
운영에서는 verify_certs=false를 사용하지 않습니다.
환경별 index 이름을 분리합니다. 예: rag_docs_dev, rag_docs_stg, rag_docs_prod.

클라이언트 구현 #

client.py

from opensearchpy import OpenSearch

from app.settings import OpenSearchSettings

def create_opensearch_client(settings: OpenSearchSettings | None = None) -> OpenSearch:
    settings = settings or OpenSearchSettings()

    return OpenSearch(
        hosts=[{"host": settings.host, "port": settings.port}],
        http_auth=(settings.username, settings.password),
        use_ssl=settings.use_ssl,
        verify_certs=settings.verify_certs,
        ssl_show_warn=not settings.verify_certs,
        timeout=30,
        max_retries=3,
        retry_on_timeout=True,
    )

from opensearchpy import OpenSearch

from app.settings import OpenSearchSettings

def create_opensearch_client(settings: OpenSearchSettings | None = None) -> OpenSearch:
    settings = settings or OpenSearchSettings()

    return OpenSearch(
        hosts=[{"host": settings.host, "port": settings.port}],
        http_auth=(settings.username, settings.password),
        use_ssl=settings.use_ssl,
        verify_certs=settings.verify_certs,
        ssl_show_warn=not settings.verify_certs,
        timeout=30,
        max_retries=3,
        retry_on_timeout=True,
    )

client.py는 실제 OpenSearch 연결 객체를 만드는 파일입니다.

create_opensearch_client() 함수는 OpenSearch 서버와 통신할 수 있는 클라이언트를 생성합니다.
settings 값이 따로 전달되지 않으면 OpenSearchSettings()를 새로 만들어 .env 설정을 사용합니다.
hosts에는 접속할 OpenSearch 서버 주소와 포트가 들어갑니다.
http_auth에는 사용자 이름과 비밀번호가 들어갑니다.
use_ssl, verify_certs는 HTTPS 연결과 인증서 검증 여부를 결정합니다.
timeout, max_retries, retry_on_timeout은 연결이 느리거나 일시적으로 실패할 때를 대비한 옵션입니다.

즉, client.py는 “OpenSearch와 대화할 수 있는 연결 도구를 만드는 파일”입니다.

healthcheck.py

from app.client import create_opensearch_client

def main() -> None:
    client = create_opensearch_client()

    info = client.info()
    print("cluster_name:", info.get("cluster_name"))
    print("version:", info.get("version", {}).get("number"))

    health = client.cluster.health()
    print("status:", health.get("status"))
    print("number_of_nodes:", health.get("number_of_nodes"))

if __name__ == "__main__":
    main()

from app.client import create_opensearch_client

def main() -> None:
    client = create_opensearch_client()

    info = client.info()
    print("cluster_name:", info.get("cluster_name"))
    print("version:", info.get("version", {}).get("number"))

    health = client.cluster.health()
    print("status:", health.get("status"))
    print("number_of_nodes:", health.get("number_of_nodes"))

if __name__ == "__main__":
    main()

healthcheck.py는 연결이 잘 되는지 확인하는 실행 파일입니다.

create_opensearch_client()로 OpenSearch 클라이언트를 만듭니다.
client.info()는 OpenSearch 서버의 기본 정보를 가져옵니다.
client.cluster.health()는 클러스터 상태를 가져옵니다.

실행 #

python -m app.healthcheck

python -m app.healthcheck

기대 결과

<https://localhost:9200> using SSL with verify_certs=False is insecure.
  warnings.warn(
c:\\Users\\user\\Documents\\project\\OpenSearch\\.venv\\Lib\\site-packages\\urllib3\\connectionpool.py:1097: InsecureRequestWarning: Unverified HTTPS request is being made to host 'localhost'. Adding certificate verification is strongly advised. See: <https://urllib3.readthedocs.io/en/latest/advanced-usage.html#tls-warnings>
  warnings.warn(
cluster_name: docker-cluster
version: 3.6.0
c:\\Users\\user\\Documents\\project\\OpenSearch\\.venv\\Lib\\site-packages\\urllib3\\connectionpool.py:1097: InsecureRequestWarning: Unverified HTTPS request is being made to host 'localhost'. Adding certificate verification is strongly advised. See: <https://urllib3.readthedocs.io/en/latest/advanced-usage.html#tls-warnings>
  warnings.warn(
status: yellow
number_of_nodes: 1

<https://localhost:9200> using SSL with verify_certs=False is insecure.
  warnings.warn(
c:\\Users\\user\\Documents\\project\\OpenSearch\\.venv\\Lib\\site-packages\\urllib3\\connectionpool.py:1097: InsecureRequestWarning: Unverified HTTPS request is being made to host 'localhost'. Adding certificate verification is strongly advised. See: <https://urllib3.readthedocs.io/en/latest/advanced-usage.html#tls-warnings>
  warnings.warn(
cluster_name: docker-cluster
version: 3.6.0
c:\\Users\\user\\Documents\\project\\OpenSearch\\.venv\\Lib\\site-packages\\urllib3\\connectionpool.py:1097: InsecureRequestWarning: Unverified HTTPS request is being made to host 'localhost'. Adding certificate verification is strongly advised. See: <https://urllib3.readthedocs.io/en/latest/advanced-usage.html#tls-warnings>
  warnings.warn(
status: yellow
number_of_nodes: 1

단일 노드 로컬 환경에서 replica가 0이면 green, replica가 1 이상인데 노드가 하나뿐이면 yellow가 될 수 있습니다.

실습 환경에서 yellow 자체가 항상 장애를 의미하지는 않지만, 원인을 이해해야 합니다.

예외 처리 고려

from opensearchpy import OpenSearchException

from app.client import create_opensearch_client

def check_connection() -> bool:
    client = create_opensearch_client()
    try:
        return bool(client.ping())
    except OpenSearchException as exc:
        print(f"OpenSearch connection failed: {exc}")
        return False

from opensearchpy import OpenSearchException

from app.client import create_opensearch_client

def check_connection() -> bool:
    client = create_opensearch_client()
    try:
        return bool(client.ping())
    except OpenSearchException as exc:
        print(f"OpenSearch connection failed: {exc}")
        return False

예외처리는 코드 실행 중 문제가 생겼을 때 프로그램이 바로 중단되지 않도록 처리하는 방식입니다.

OpenSearch 연결에서는 다음과 같은 문제가 자주 발생할 수 있습니다.

OpenSearch 컨테이너가 실행 중이 아님
포트 번호가 잘못됨
아이디 또는 비밀번호가 틀림
SSL 설정이 맞지 않음
네트워크 연결이 불안정함
인증서 검증 설정이 맞지 않음

이런 상황에서 예외처리를 하지 않으면 Python 프로그램은 에러 메시지를 출력하고 바로 종료됩니다. 하지만 try-except를 사용하면 실패 상황을 직접 처리할 수 있습니다.

운영 서비스에서 추가로 고려할 client 옵션 #

옵션	의미	권장 방향
`timeout`	요청 제한 시간	검색 API와 bulk indexing API를 구분해 설정
`max_retries`	재시도 횟수	일시적 네트워크 오류에 대비
`retry_on_timeout`	timeout 발생 시 재시도	읽기 요청 중심으로 신중히 사용
`verify_certs`	TLS 인증서 검증	운영에서는 반드시 true

AI 코딩 에이전트 Wiki

LangChain/LangGraph

OpenSearch

Python 개발 환경 구성

권장 프로젝트 구조 #

가상환경 생성 #

패키지 설치 #

설정 #

설정 관리 원칙 #

클라이언트 구현 #

실행 #

운영 서비스에서 추가로 고려할 client 옵션 #

글이 도움이 되셨나요?

답글 남기기 응답 취소

권장 프로젝트 구조 #

가상환경 생성 #

패키지 설치 #

설정 #

설정 관리 원칙 #

클라이언트 구현 #

실행 #

운영 서비스에서 추가로 고려할 client 옵션 #

글이 도움이 되셨나요?

공유하기

답글 남기기 응답 취소