AgentCoreService/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'

def get():
    dirname = 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').")
    return yaml.safe_load(open(f'{dirname}/{env}.yaml').read())

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())