hgzero/rag/src/utils/config.py
2025-10-29 05:54:08 +09:00

120 lines
2.9 KiB
Python

"""
설정 관리 유틸리티
"""
import os
import yaml
from typing import Any, Dict
from pathlib import Path
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
"""애플리케이션 설정"""
# PostgreSQL
POSTGRES_HOST: str = "localhost"
POSTGRES_PORT: int = 5432
POSTGRES_DATABASE: str = "meeting_db"
POSTGRES_USER: str = "postgres"
POSTGRES_PASSWORD: str = ""
# Azure OpenAI
AZURE_OPENAI_API_KEY: str = ""
AZURE_OPENAI_ENDPOINT: str = ""
# Azure AI Search
AZURE_SEARCH_ENDPOINT: str = ""
AZURE_SEARCH_API_KEY: str = ""
# Claude AI
CLAUDE_API_KEY: str = ""
# Redis
REDIS_PASSWORD: str = ""
# Azure Event Hub
EVENTHUB_CONNECTION_STRING: str = ""
EVENTHUB_NAME: str = ""
AZURE_EVENTHUB_CONSUMER_GROUP: str = "$Default"
AZURE_STORAGE_CONNECTION_STRING: str = ""
AZURE_STORAGE_CONTAINER_NAME: str = ""
class Config:
# rag 디렉토리 기준으로 .env 파일 경로 설정
env_file = str(Path(__file__).parent.parent.parent / ".env")
case_sensitive = True
def load_config(config_path: str = "config.yaml") -> Dict[str, Any]:
"""
설정 파일 로딩
Args:
config_path: 설정 파일 경로
Returns:
설정 딕셔너리
"""
# 환경변수 로딩
settings = Settings()
# YAML 파일 로딩
config_file = Path(config_path)
if not config_file.exists():
raise FileNotFoundError(f"설정 파일을 찾을 수 없습니다: {config_path}")
with open(config_file, "r", encoding="utf-8") as f:
config = yaml.safe_load(f)
# 환경변수로 대체
def replace_env_vars(obj: Any) -> Any:
"""재귀적으로 환경변수 치환"""
if isinstance(obj, dict):
return {k: replace_env_vars(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [replace_env_vars(item) for item in obj]
elif isinstance(obj, str) and obj.startswith("${") and obj.endswith("}"):
env_var = obj[2:-1]
return getattr(settings, env_var, "")
return obj
config = replace_env_vars(config)
return config
def get_database_url(config: Dict[str, Any]) -> str:
"""
PostgreSQL 데이터베이스 URL 생성
Args:
config: 설정 딕셔너리
Returns:
데이터베이스 URL
"""
pg = config["postgres"]
return (
f"postgresql://{pg['user']}:{pg['password']}"
f"@{pg['host']}:{pg['port']}/{pg['database']}"
)
def get_redis_url(config: Dict[str, Any]) -> str:
"""
Redis URL 생성
Args:
config: 설정 딕셔너리
Returns:
Redis URL
"""
redis = config["redis"]
password = redis.get("password", "")
if password:
return f"redis://:{password}@{redis['host']}:{redis['port']}/{redis['db']}"
else:
return f"redis://{redis['host']}:{redis['port']}/{redis['db']}"