nb_config_center

A Redis-based distributed configuration center for Python applications with real-time synchronization and type preservation.

Features

• Centralized Configuration Storage: Store configuration as key-value pairs in Redis Hash
• Type Preservation: Automatically preserves Python types (int, float, bool, dict, list, str) using JSON serialization
• Real-time Synchronization: Uses Redis Pub/Sub to notify all instances when configuration changes
• Callback Support: Register callbacks to execute custom logic when configuration updates
• Resource Efficiency: Reuses Redis connections and subscriber threads across multiple instances
• Thread-Safe: Built-in thread safety for concurrent access

Installation

pip install nb_config_center

Requirements

• Python >= 3.7
• redis

Quick Start

import redis
from nb_config_center import NbConfigCenter

# Create Redis client
r = redis.StrictRedis(host='localhost', port=6379, db=0, decode_responses=True)

# Initialize configuration center
config = NbConfigCenter(r, namespace="my_app_config")

# Update configuration
config.update_config({
    "database_url": "postgresql://localhost/mydb",
    "max_connections": 100,
    "debug_mode": True,
    "features": {"feature_a": True, "feature_b": False}
})

# Get configuration value
max_conn = config.get("max_connections")  # Returns: 100 (int)

Usage Examples

Basic Configuration Management

import redis
from nb_config_center import NbConfigCenter

r = redis.StrictRedis(host='localhost', port=6379, db=0, decode_responses=True)
config = NbConfigCenter(r, namespace="app_config")

# Update configuration
config.update_config({
    "api_key": "secret123",
    "timeout": 30,
    "retry_enabled": True
})

# Access configuration
print(config.get("timeout"))  # 30
print(config.config)  # {'api_key': 'secret123', 'timeout': 30, 'retry_enabled': True}

Multiple Instances with Real-time Sync

import redis
from nb_config_center import NbConfigCenter
import time

r = redis.StrictRedis(host='localhost', port=6379, db=0, decode_responses=True)

# Instance 1: Updates configuration
config1 = NbConfigCenter(r, namespace="shared_config")
config1.update_config({"version": "1.0.0", "max_users": 1000})

# Instance 2: Automatically receives updates via Redis Pub/Sub
config2 = NbConfigCenter(r, namespace="shared_config")

# Register callback for configuration changes
@config2.add_update_callback
def on_config_change(old_config, new_config):
    print(f"Configuration updated!")
    print(f"Old: {old_config}")
    print(f"New: {new_config}")

# When config1 updates, config2's callback will be triggered
config1.update_config({"max_users": 2000})
# Output: Configuration updated! Old: {...} New: {'version': '1.0.0', 'max_users': 2000}

Type Preservation

import redis
from nb_config_center import NbConfigCenter

r = redis.StrictRedis(host='localhost', port=6379, db=0, decode_responses=True)
config = NbConfigCenter(r, namespace="typed_config")

# Store different types
config.update_config({
    "string_val": "hello",
    "int_val": 42,
    "float_val": 3.14,
    "bool_val": True,
    "list_val": [1, 2, 3],
    "dict_val": {"nested": "value"}
})

# Types are preserved when retrieved
print(type(config.get("int_val")))    # <class 'int'>
print(type(config.get("float_val")))  # <class 'float'>
print(type(config.get("bool_val")))   # <class 'bool'>
print(type(config.get("list_val")))   # <class 'list'>
print(type(config.get("dict_val")))   # <class 'dict'>

Using Custom Channel Names

When multiple projects share the same Redis instance, use different channel names to avoid cross-talk:

import redis
from nb_config_center import NbConfigCenter

r = redis.StrictRedis(host='localhost', port=6379, db=0, decode_responses=True)

# Project A
config_a = NbConfigCenter(r, namespace="project_a", channel_name="project_a_updates")

# Project B
config_b = NbConfigCenter(r, namespace="project_b", channel_name="project_b_updates")

# Updates to config_a won't trigger callbacks in config_b

API Reference

NbConfigCenter

Main class for managing distributed configuration.

Constructor

NbConfigCenter(redis_client, namespace: str, channel_name: str = "nb_config_update_bus")

Parameters:

• redis_client: Redis client object (from redis package)
• namespace: Configuration namespace (used as Redis Hash key)
• channel_name: Redis Pub/Sub channel name for notifications (default: "nb_config_update_bus")

Methods

get(key: str)

Retrieve a configuration value by key.

Parameters:

• key: Configuration key

Returns: Configuration value with original type preserved

Raises: KeyError if key doesn't exist

update_config(config: dict)

Update configuration to Redis and notify all instances.

Parameters:

• config: Dictionary of configuration key-value pairs to update

Example:

config.update_config({"key1": "value1", "key2": 123})

add_update_callback(func)

Register a callback function to be executed when configuration changes. Can be used as a decorator.

Parameters:

• func: Callback function that accepts two parameters: old_config and new_config

Returns: The function (allows use as decorator)

Example:

@config.add_update_callback
def my_callback(old_config, new_config):
    print(f"Config changed from {old_config} to {new_config}")

Attributes

• config: Current configuration dictionary
• old_config: Previous configuration dictionary (before last update)
• namespace: Configuration namespace
• redis: Redis client instance

How It Works

1. Storage: Configuration is stored in Redis as a Hash, with each key-value pair serialized using JSON to preserve types
2. Synchronization: When update_config() is called, the changes are:
   • Written to Redis Hash atomically
   • Published to a Redis Pub/Sub channel
3. Notification: All NbConfigCenter instances subscribed to the same namespace receive the notification
4. Refresh: Each instance pulls the latest configuration from Redis
5. Callbacks: Registered callbacks are executed with old and new configuration

Architecture

┌─────────────┐         ┌─────────────┐         ┌─────────────┐
│  Instance 1 │         │  Instance 2 │         │  Instance N │
│             │         │             │         │             │
│  update()   │         │  callback() │         │  callback() │
└──────┬──────┘         └──────▲──────┘         └──────▲──────┘
       │                       │                        │
       │ 1. Write Hash         │ 3. Receive Pub/Sub    │
       │ 2. Publish            │ 4. Refresh from Hash  │
       ▼                       │                        │
┌─────────────────────────────────────────────────────────────┐
│                         Redis Server                        │
│  ┌──────────────┐              ┌──────────────────────┐    │
│  │  Hash Store  │              │  Pub/Sub Channel     │    │
│  │  namespace:  │              │  nb_config_update_bus│    │
│  │  {k1: v1...} │              │                      │    │
│  └──────────────┘              └──────────────────────┘    │
└─────────────────────────────────────────────────────────────┘

Best Practices

1. Namespace Isolation: Use unique namespaces for different applications or environments
2. Channel Names: When sharing Redis across projects, use unique channel names
3. Callback Design: Keep callbacks lightweight and avoid blocking operations
4. Error Handling: Callbacks should handle their own exceptions to prevent disrupting the update process
5. Initial Load: Callbacks are NOT triggered on the first load when an instance is created

License

MIT License - see LICENSE file for details

Links

• GitHub: https://github.com/ydf0509/nb_config_center
• Issues: https://github.com/ydf0509/nb_config_center/issues
• PyPI: https://pypi.org/project/nb_config_center/

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.