CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-nacos-sdk-python

A Python implementation of Nacos OpenAPI for service discovery, configuration management, and service management

—

Pending
Overview
Eval results
Files

v1-config.mddocs/

V1 Configuration Management (Synchronous)

Legacy synchronous API for Nacos configuration operations. This API provides basic configuration management functionality using HTTP requests with threading-based operations for backward compatibility.

Capabilities

Client Initialization

Create a synchronous Nacos client for configuration operations.

class NacosClient:
    def __init__(self, server_addresses=None, endpoint=None, namespace=None, ak=None, sk=None, 
                 username=None, password=None, app_name=None, app_key=None, log_dir=None, 
                 log_level=None, log_rotation_backup_count=None, **kwargs):
        """
        Initialize Nacos client.
        
        Args:
            server_addresses (str): Comma-separated list of Nacos server addresses
            endpoint (str): Address server endpoint for dynamic server discovery
            namespace (str): Nacos namespace ID
            ak (str): Access key for authentication
            sk (str): Secret key for authentication  
            username (str): Username for authentication
            password (str): Password for authentication
            app_name (str): Application name
            app_key (str): Application key
            log_dir (str): Directory for log files
            log_level: Logging level
            log_rotation_backup_count (int): Number of backup log files
        """

Usage example:

from nacos import NacosClient

# Basic client with server addresses
client = NacosClient(server_addresses="127.0.0.1:8848,127.0.0.1:8849")

# Client with authentication
client = NacosClient(
    server_addresses="127.0.0.1:8848",
    namespace="production",
    username="nacos",
    password="nacos"
)

# Client with access key authentication
client = NacosClient(
    server_addresses="127.0.0.1:8848",
    ak="your-access-key",
    sk="your-secret-key"
)

Get Configuration

Retrieve configuration values from Nacos server.

def get_config(self, data_id: str, group: str, timeout=None, no_snapshot=None) -> str:
    """
    Get configuration from Nacos server.
    
    Args:
        data_id (str): Configuration data ID
        group (str): Configuration group
        timeout (int, optional): Request timeout in seconds
        no_snapshot (bool, optional): Disable local snapshot fallback
        
    Returns:
        str: Configuration content
        
    Raises:
        NacosException: If data_id or group is invalid
        NacosRequestException: If request fails
    """

Usage example:

# Get configuration
config = client.get_config("database.config", "DEFAULT_GROUP")
print(config)

# Get with custom timeout
config = client.get_config("database.config", "DEFAULT_GROUP", timeout=10)

# Get with snapshot disabled
config = client.get_config("database.config", "DEFAULT_GROUP", no_snapshot=True)

Publish Configuration

Publish or update configuration values on Nacos server.

def publish_config(self, data_id: str, group: str, content: str, app_name=None, 
                  config_type=None, timeout=None) -> bool:
    """
    Publish configuration to Nacos server.
    
    Args:
        data_id (str): Configuration data ID
        group (str): Configuration group
        content (str): Configuration content
        app_name (str, optional): Application name
        config_type (str, optional): Configuration type (e.g., 'json', 'yaml', 'properties')
        timeout (int, optional): Request timeout in seconds
        
    Returns:
        bool: True if successful
        
    Raises:
        NacosException: If data_id or group is invalid
        NacosRequestException: If request fails
    """

Usage example:

# Publish simple configuration
client.publish_config("database.config", "DEFAULT_GROUP", "host=localhost\nport=3306")

# Publish JSON configuration
config_content = '{"host": "localhost", "port": 3306}'
client.publish_config("database.config", "DEFAULT_GROUP", config_content, config_type="json")

# Publish with app name
client.publish_config("database.config", "DEFAULT_GROUP", config_content, app_name="my-app")

Remove Configuration

Remove configuration from Nacos server.

def remove_config(self, data_id: str, group: str, timeout=None) -> bool:
    """
    Remove configuration from Nacos server.
    
    Args:
        data_id (str): Configuration data ID
        group (str): Configuration group
        timeout (int, optional): Request timeout in seconds
        
    Returns:
        bool: True if successful
        
    Raises:
        NacosException: If data_id or group is invalid
        NacosRequestException: If request fails
    """

Usage example:

# Remove configuration
success = client.remove_config("database.config", "DEFAULT_GROUP")
if success:
    print("Configuration removed successfully")

Configuration Watchers

Add and manage configuration change listeners.

def add_config_watcher(self, data_id: str, group: str, cb, content=None):
    """
    Add configuration change listener.
    
    Args:
        data_id (str): Configuration data ID
        group (str): Configuration group
        cb (callable): Callback function to handle configuration changes
        content (str, optional): Initial content for comparison
    """

def add_config_watchers(self, data_id: str, group: str, cb_list, content=None):
    """
    Add multiple configuration change listeners.
    
    Args:
        data_id (str): Configuration data ID
        group (str): Configuration group
        cb_list (list): List of callback functions
        content (str, optional): Initial content for comparison
    """

def remove_config_watcher(self, data_id: str, group: str, cb, remove_all=False):
    """
    Remove configuration change listener.
    
    Args:
        data_id (str): Configuration data ID
        group (str): Configuration group
        cb (callable): Callback function to remove
        remove_all (bool): Remove all listeners for this config
    """

Usage example:

def config_change_handler(args):
    print(f"Configuration changed: {args}")
    # Handle configuration change
    
# Add single listener
client.add_config_watcher("database.config", "DEFAULT_GROUP", config_change_handler)

# Add multiple listeners
def backup_handler(args):
    print(f"Backup handler: {args}")
    
client.add_config_watchers("database.config", "DEFAULT_GROUP", 
                          [config_change_handler, backup_handler])

# Remove specific listener
client.remove_config_watcher("database.config", "DEFAULT_GROUP", config_change_handler)

# Remove all listeners
client.remove_config_watcher("database.config", "DEFAULT_GROUP", None, remove_all=True)

Batch Configuration Operations

Retrieve multiple configurations in a single request.

def get_configs(self, timeout=None, no_snapshot=None, group="", page_no=1, page_size=1000) -> dict:
    """
    Get multiple configurations with pagination.
    
    Args:
        timeout (int, optional): Request timeout in seconds
        no_snapshot (bool, optional): Disable local snapshot fallback
        group (str): Configuration group filter
        page_no (int): Page number (1-based)
        page_size (int): Number of configurations per page
        
    Returns:
        dict: Dictionary containing configurations and pagination info
    """

Usage example:

# Get all configurations
configs = client.get_configs()
print(f"Total configs: {configs.get('totalCount', 0)}")

# Get configurations with pagination
configs = client.get_configs(page_no=1, page_size=50)
for config in configs.get('pageItems', []):
    print(f"Config: {config['dataId']} = {config['content']}")

# Get configurations from specific group
configs = client.get_configs(group="DEFAULT_GROUP")

Error Handling

class NacosException(Exception):
    """Base exception for Nacos operations."""
    pass

class NacosRequestException(NacosException):
    """Exception for HTTP request failures."""
    pass

Common error scenarios:

from nacos import NacosClient, NacosException, NacosRequestException

try:
    client = NacosClient(server_addresses="127.0.0.1:8848")
    config = client.get_config("test-config", "DEFAULT_GROUP")
except NacosRequestException as e:
    print(f"Request failed: {e}")
except NacosException as e:
    print(f"Nacos error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

Constants and Defaults

DEFAULT_GROUP_NAME = "DEFAULT_GROUP"
DEFAULTS = {
    "APP_NAME": "Nacos-SDK-Python",
    "TIMEOUT": 3,
    "PULLING_TIMEOUT": 30,
    "PULLING_CONFIG_SIZE": 3000,
    "CALLBACK_THREAD_NUM": 10,
    "FAILOVER_BASE": "nacos-data/data",
    "SNAPSHOT_BASE": "nacos-data/snapshot"
}

Install with Tessl CLI

npx tessl i tessl/pypi-nacos-sdk-python

docs

client-config.md

index.md

v1-config.md

v1-naming.md

v2-config.md

v2-naming.md

tile.json