ask_candid/base/config/connections.py

from dataclasses import dataclass, field
import os

from dotenv import dotenv_values, find_dotenv


@dataclass
class BaseElasticSearchConnection:
    """Elasticsearch connection dataclass
    """
    url: str = field(default_factory=str)
    username: str = field(default_factory=str)
    password: str = field(default_factory=str)


@dataclass
class BaseElasticAPIKeyCredential:
    """Cloud ID/API key data class
    """
    cloud_id: str = field(default_factory=str)
    api_key: str = field(default_factory=str)


__env_values__ = dotenv_values(
    dotenv_path=find_dotenv(".env", raise_error_if_not_found=False)
)

def _load_value(key: str):
    return __env_values__.get(key) or os.getenv(key)

SEMANTIC_ELASTIC_QA = BaseElasticAPIKeyCredential(
    cloud_id=_load_value("SEMANTIC_ELASTIC_CLOUD_ID"),
    api_key=_load_value("SEMANTIC_ELASTIC_API_KEY"),
)

SEMANTIC_ELASTIC_QA_WRITER = BaseElasticAPIKeyCredential(
    cloud_id=_load_value("SEMANTIC_ELASTIC_WRITER_CLOUD_ID"),
    api_key=_load_value("SEMANTIC_ELASTIC_WRITER_API_KEY"),
)

NEWS_ELASTIC = BaseElasticSearchConnection(
    url=_load_value("NEWS_URL"),
    username=_load_value("NEWS_UID"),
    password=_load_value("NEWS_PWD")
)