使用Python实现获取Apollo配置

pip install apollo-client

from apollo.client import ApolloClient

# 创建Apollo客户端实例
client = ApolloClient(
    app_id="your-app-id",
    cluster="default",
    config_server_url="http://your-apollo-server:8080"
)

# 获取默认namespace(application)的所有配置
config = client.get_value()
print(config)  # 输出所有配置

# 获取特定key的值
db_url = client.get_value("mysql.url")
print(f"数据库URL: {db_url}")

# 获取指定namespace的所有配置
db_config = client.get_namespace("database")
print(db_config)  # 输出database namespace的所有配置

# 获取指定namespace中特定key的值
redis_host = client.get_value("redis.host", namespace="redis-config")
print(f"Redis主机地址: {redis_host}")

def config_change_handler(changes):
    for key, value in changes.items():
        print(f"配置变更: {key} = {value}")

# 注册配置变更监听器
client.start_listening(config_change_handler)

# 同时使用多个namespace
client.get_namespace("application")  # 默认namespace
client.get_namespace("database")     # 数据库配置
client.get_namespace("redis")        # Redis配置

# 强制刷新缓存
client.refresh_namespace("application")

# 获取本地缓存的配置
cached_config = client.get_cached_value("mysql.url")

try:
    config = client.get_value("non-existent-key")
except Exception as e:
    print(f"获取配置失败: {e}")
    # 使用默认值
    config = "default_value"

class ConfigService:
    def __init__(self):
        self.client = ApolloClient(
            app_id="your-app-id",
            cluster="default",
            config_server_url="http://your-apollo-server:8080"
        )
    
    def get_database_config(self):
        return {
            "host": self.client.get_value("mysql.host", namespace="database"),
            "port": self.client.get_value("mysql.port", namespace="database"),
            "username": self.client.get_value("mysql.username", namespace="database"),
            "password": self.client.get_value("mysql.password", namespace="database")
        }
    
    def get_redis_config(self):
        return {
            "host": self.client.get_value("redis.host", namespace="redis"),
            "port": self.client.get_value("redis.port", namespace="redis")
        }

# 使用示例
config_service = ConfigService()
db_config = config_service.get_database_config()
redis_config = config_service.get_redis_config()

def validate_db_config(config):
    required_fields = ["host", "port", "username", "password"]
    for field in required_fields:
        if not config.get(field):
            raise ValueError(f"数据库配置缺少必要字段: {field}")
    
    if not isinstance(config["port"], int):
        raise ValueError("数据库端口必须是整数")

# 使用示例
try:
    db_config = config_service.get_database_config()
    validate_db_config(db_config)
except ValueError as e:
    print(f"配置验证失败: {e}")

client = ApolloClient(
    app_id="your-app-id",
    cluster="default",
    config_server_url="http://your-apollo-server:8080",
    timeout=5  # 设置5秒超时
)

client = ApolloClient(
    app_id="your-app-id",
    cluster="default",
    config_server_url="http://your-apollo-server:8080",
    refresh_interval=30  # 设置30秒刷新间隔
)