#! /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())