AgentCoreService/pqai_agent/configs/__init__.py

#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vim:fenc=utf-8
import json
import os
import yaml
import pyapollos

APOLLO_PROJECT_NAME = 'ai-agent'

_config_cache = None

def get():
    global _config_cache
    if _config_cache is None:
        dir_name = os.path.dirname(os.path.abspath(__file__))
        env = os.environ.get('AI_AGENT_ENV', 'dev')
        if env not in ('dev', 'pre', 'prod'):
            raise ValueError(f"Invalid environment: {env}. Expected one of ('dev', 'pre', 'prod').")
        with open(f'{dir_name}/{env}.yaml', 'r') as f:
            _config_cache = yaml.safe_load(f.read())
    return _config_cache

def get_env():
    env = os.environ.get('AI_AGENT_ENV', 'dev')
    return env

class ApolloConfig(object):
    def __init__(self, env):
        match env:
            case "dev":
                self.apollo_connection = pyapollos.ApolloClient(
                    app_id=APOLLO_PROJECT_NAME,
                    config_server_url="http://devapolloconfig-internal.piaoquantv.com/",
                    timeout=10
                )
            case "pre":
                self.apollo_connection = pyapollos.ApolloClient(
                    app_id=APOLLO_PROJECT_NAME,
                    config_server_url="http://preapolloconfig-internal.piaoquantv.com/",
                    timeout=10
                )
            case "prod":
                self.apollo_connection = pyapollos.ApolloClient(
                    app_id=APOLLO_PROJECT_NAME,
                    config_server_url="https://apolloconfig-internal.piaoquantv.com/",
                    timeout=10
                )

    def get_value(self, key, default=None):
        response = self.apollo_connection.get_value(key, default)
        return response

    def get_json_value(self, key, default=None):
        value = self.get_value(key)
        if not value:
            return default
        else:
            return json.loads(value)

apollo_config = ApolloConfig(get_env())