A Python implementation of Nacos OpenAPI for service discovery, configuration management, and service management
—
Configuration classes for setting up Nacos clients with support for authentication, security, networking, logging, and operational parameters. The V2 API provides comprehensive configuration options through builder patterns and specialized configuration classes.
Main configuration class for Nacos client initialization.
class ClientConfig:
def __init__(self, server_addresses=None, endpoint=None, namespace_id='', context_path='',
access_key=None, secret_key=None, username=None, password=None, app_name='',
app_key='', log_dir='', log_level=None, log_rotation_backup_count=None,
app_conn_labels=None, credentials_provider=None):
"""
Initialize client configuration.
Args:
server_addresses (str): Comma-separated list of Nacos server addresses
endpoint (str): Address server endpoint for dynamic server discovery
namespace_id (str): Nacos namespace ID
context_path (str): Custom context path for Nacos server
access_key (str): Access key for authentication
secret_key (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 (logging.INFO, logging.DEBUG, etc.)
log_rotation_backup_count (int): Number of backup log files
app_conn_labels (dict): Application connection labels
credentials_provider: Custom credentials provider
"""Usage example:
from v2.nacos import ClientConfig
import logging
# Basic configuration with server addresses
client_config = ClientConfig(
server_addresses="127.0.0.1:8848,127.0.0.1:8849,127.0.0.1:8850"
)
# Configuration with authentication
client_config = ClientConfig(
server_addresses="127.0.0.1:8848",
namespace_id="production",
username="nacos",
password="nacos"
)
# Configuration with access key authentication
client_config = ClientConfig(
server_addresses="127.0.0.1:8848",
access_key="your-access-key",
secret_key="your-secret-key",
namespace_id="production"
)
# Advanced configuration
client_config = ClientConfig(
server_addresses="127.0.0.1:8848",
namespace_id="production",
username="nacos",
password="nacos",
app_name="MyApplication",
log_dir="/var/log/nacos",
log_level=logging.DEBUG,
log_rotation_backup_count=10
)Builder class for creating ClientConfig instances with fluent interface.
class ClientConfigBuilder:
def server_address(self, server_address: str) -> 'ClientConfigBuilder': ...
def endpoint(self, endpoint: str) -> 'ClientConfigBuilder': ...
def namespace_id(self, namespace_id: str) -> 'ClientConfigBuilder': ...
def context_path(self, context_path: str) -> 'ClientConfigBuilder': ...
def access_key(self, access_key: str) -> 'ClientConfigBuilder': ...
def secret_key(self, secret_key: str) -> 'ClientConfigBuilder': ...
def username(self, username: str) -> 'ClientConfigBuilder': ...
def password(self, password: str) -> 'ClientConfigBuilder': ...
def app_name(self, app_name: str) -> 'ClientConfigBuilder': ...
def app_key(self, app_key: str) -> 'ClientConfigBuilder': ...
def log_dir(self, log_dir: str) -> 'ClientConfigBuilder': ...
def log_level(self, log_level) -> 'ClientConfigBuilder': ...
def log_rotation_backup_count(self, count: int) -> 'ClientConfigBuilder': ...
def build(self) -> ClientConfig: ...Usage example:
from v2.nacos import ClientConfigBuilder
import logging
# Build configuration using fluent interface
client_config = (ClientConfigBuilder()
.server_address("127.0.0.1:8848,127.0.0.1:8849")
.namespace_id("production")
.username("nacos")
.password("nacos")
.app_name("MyApplication")
.log_level(logging.INFO)
.build())
# Build step by step
builder = ClientConfigBuilder()
builder.server_address("127.0.0.1:8848")
builder.namespace_id("staging")
builder.access_key("staging-access-key")
builder.secret_key("staging-secret-key")
builder.app_name("StagingApp")
client_config = builder.build()Fluent methods for updating configuration after creation.
class ClientConfig:
def set_log_level(self, log_level): ...
def set_cache_dir(self, cache_dir): ...
def set_log_dir(self, log_dir): ...
def set_timeout_ms(self, timeout_ms): ...
def set_heart_beat_interval(self, heart_beat_interval): ...
def set_kms_config(self, kms_config): ...
def set_tls_config(self, tls_config): ...
def set_grpc_config(self, grpc_config): ...Usage example:
import logging
# Create basic config
client_config = ClientConfig(server_addresses="127.0.0.1:8848")
# Update configuration
client_config.set_log_level(logging.DEBUG)
client_config.set_cache_dir("/tmp/nacos-cache")
client_config.set_timeout_ms(15000) # 15 seconds
client_config.set_heart_beat_interval(3000) # 3 seconds
# Chain method calls
client_config = (ClientConfig(server_addresses="127.0.0.1:8848")
.set_log_level(logging.INFO)
.set_timeout_ms(10000)
.set_heart_beat_interval(5000))Configuration for Key Management Service encryption support.
class KMSConfig:
def __init__(self, enabled=False, endpoint='', access_key='', secret_key='',
client_key_content='', password=''):
"""
Initialize KMS configuration.
Args:
enabled (bool): Whether KMS encryption is enabled
endpoint (str): KMS service endpoint URL
access_key (str): KMS access key
secret_key (str): KMS secret key
client_key_content (str): Client key content for authentication
password (str): Password for key authentication
"""Usage example:
from v2.nacos import ClientConfig, KMSConfig
# Configure KMS encryption
kms_config = KMSConfig(
enabled=True,
endpoint="https://kms.us-west-1.amazonaws.com",
access_key="AKIAIOSFODNN7EXAMPLE",
secret_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
)
# Use KMS config in client configuration
client_config = ClientConfig(
server_addresses="127.0.0.1:8848",
kms_config=kms_config
)
# Alternative: Set KMS config after creation
client_config = ClientConfig(server_addresses="127.0.0.1:8848")
client_config.set_kms_config(kms_config)Configuration for secure TLS/SSL connections.
class TLSConfig:
def __init__(self, enabled=False, appointed=False, ca_file='', cert_file='',
key_file='', server_name_override=''):
"""
Initialize TLS configuration.
Args:
enabled (bool): Whether TLS is enabled
appointed (bool): Whether to use predefined configuration
ca_file (str): Path to CA certificate file
cert_file (str): Path to client certificate file
key_file (str): Path to private key file
server_name_override (str): Server name override for testing
"""Usage example:
from v2.nacos import ClientConfig, TLSConfig
# Configure TLS with certificates
tls_config = TLSConfig(
enabled=True,
ca_file="/etc/ssl/certs/ca-cert.pem",
cert_file="/etc/ssl/certs/client-cert.pem",
key_file="/etc/ssl/private/client-key.pem"
)
# Use TLS config in client configuration
client_config = ClientConfig(
server_addresses="https://nacos.example.com:8848",
tls_config=tls_config
)
# TLS with server name override (for testing)
tls_config = TLSConfig(
enabled=True,
appointed=True,
server_name_override="localhost"
)
client_config = ClientConfig(
server_addresses="127.0.0.1:8848",
tls_config=tls_config
)
client_config.set_tls_config(tls_config)Configuration for GRPC connection parameters and performance tuning.
class GRPCConfig:
def __init__(self, max_receive_message_length=4194304, max_keep_alive_ms=60000,
initial_window_size=1048576, initial_conn_window_size=1048576,
grpc_timeout=5000):
"""
Initialize GRPC configuration.
Args:
max_receive_message_length (int): Maximum message size in bytes
max_keep_alive_ms (int): Maximum keep-alive time in milliseconds
initial_window_size (int): Initial window size for flow control
initial_conn_window_size (int): Initial connection window size
grpc_timeout (int): GRPC operation timeout in milliseconds
"""Usage example:
from v2.nacos import ClientConfig, GRPCConfig
# Default GRPC configuration
grpc_config = GRPCConfig()
# Custom GRPC configuration for high-throughput
grpc_config = GRPCConfig(
max_receive_message_length=16777216, # 16MB
max_keep_alive_ms=30000, # 30 seconds
initial_window_size=2097152, # 2MB
initial_conn_window_size=2097152, # 2MB
grpc_timeout=10000 # 10 seconds
)
# Use GRPC config in client configuration
client_config = ClientConfig(
server_addresses="127.0.0.1:8848",
grpc_config=grpc_config
)
# Set GRPC config after creation
client_config = ClientConfig(server_addresses="127.0.0.1:8848")
client_config.set_grpc_config(grpc_config)Additional configuration properties for fine-tuning client behavior.
class ClientConfig:
# Core connection settings
timeout_ms: int = 10000 # Request timeout in milliseconds
heart_beat_interval: int = 5000 # Heartbeat interval in milliseconds
# Directory settings
cache_dir: str = '' # Local cache directory
log_dir: str = '' # Log directory
# Logging settings
log_level: int = logging.INFO # Logging level
log_rotation_backup_count: int = 7 # Number of backup log files
# Cache settings
load_cache_at_start: bool = True # Load cache on startup
update_cache_when_empty: bool = False # Update cache when empty
disable_use_config_cache: bool = False # Disable configuration cache
# Service discovery settings
async_update_service: bool = False # Enable async service updates
update_thread_num: int = 5 # Number of update threads
# Connection settings
app_conn_labels: dict = None # Application connection labelsUsage example:
# Performance-optimized configuration
client_config = ClientConfig(
server_addresses="127.0.0.1:8848",
timeout_ms=5000, # Faster timeout
heart_beat_interval=3000, # More frequent heartbeats
async_update_service=True, # Enable async updates
update_thread_num=10 # More update threads
)
# Development configuration with extensive logging
client_config = ClientConfig(
server_addresses="127.0.0.1:8848",
log_level=logging.DEBUG,
log_rotation_backup_count=20,
cache_dir="/tmp/nacos-dev-cache",
load_cache_at_start=False, # Don't load cache on startup
disable_use_config_cache=True # Disable cache for fresh data
)
# Production configuration with caching
client_config = ClientConfig(
server_addresses="nacos1.prod.com:8848,nacos2.prod.com:8848,nacos3.prod.com:8848",
namespace_id="production",
timeout_ms=15000,
heart_beat_interval=5000,
cache_dir="/var/cache/nacos",
log_dir="/var/log/nacos",
log_level=logging.WARN,
load_cache_at_start=True,
update_cache_when_empty=True,
async_update_service=True
)from v2.nacos import ClientConfig
# Simplest possible configuration
client_config = ClientConfig(server_addresses="127.0.0.1:8848")from v2.nacos import ClientConfig, ClientConfigBuilder
import logging
# Development configuration with debugging
client_config = (ClientConfigBuilder()
.server_addresses("127.0.0.1:8848")
.namespace_id("dev")
.username("nacos")
.password("nacos")
.app_name("MyApp-Dev")
.log_level(logging.DEBUG)
.build())
# Enable development-friendly settings
client_config.set_timeout_ms(30000) # Longer timeout for debugging
client_config.set_cache_dir("/tmp/nacos-dev-cache")
client_config.disable_use_config_cache = True # Always fetch fresh datafrom v2.nacos import ClientConfig, TLSConfig, KMSConfig
import logging
# Production configuration with security
tls_config = TLSConfig(
enabled=True,
ca_file="/etc/ssl/certs/nacos-ca.pem",
cert_file="/etc/ssl/certs/app-client.pem",
key_file="/etc/ssl/private/app-client-key.pem"
)
kms_config = KMSConfig(
enabled=True,
endpoint="https://kms.us-west-1.amazonaws.com",
access_key="prod-access-key",
secret_key="prod-secret-key"
)
client_config = ClientConfig(
server_addresses="nacos1.prod.com:8848,nacos2.prod.com:8848,nacos3.prod.com:8848",
namespace_id="production",
access_key="prod-app-access-key",
secret_key="prod-app-secret-key",
app_name="MyApp-Prod",
tls_config=tls_config,
kms_config=kms_config,
log_dir="/var/log/nacos",
log_level=logging.INFO,
cache_dir="/var/cache/nacos",
timeout_ms=10000,
heart_beat_interval=5000,
async_update_service=True
)from v2.nacos import ClientConfig, GRPCConfig
# High-performance configuration
grpc_config = GRPCConfig(
max_receive_message_length=33554432, # 32MB for large configs
max_keep_alive_ms=30000,
initial_window_size=4194304, # 4MB
initial_conn_window_size=4194304, # 4MB
grpc_timeout=5000
)
client_config = ClientConfig(
server_addresses="127.0.0.1:8848",
grpc_config=grpc_config,
timeout_ms=5000,
heart_beat_interval=3000,
async_update_service=True,
update_thread_num=20, # More threads for updates
load_cache_at_start=True,
update_cache_when_empty=True
)client_config = ClientConfig(
server_addresses="127.0.0.1:8848",
username="admin",
password="password123"
)client_config = ClientConfig(
server_addresses="127.0.0.1:8848",
access_key="your-access-key",
secret_key="your-secret-key"
)from v2.nacos.common.auth import StaticCredentialsProvider
# Custom credentials provider
credentials_provider = StaticCredentialsProvider(
access_key="custom-access-key",
secret_key="custom-secret-key"
)
client_config = ClientConfig(
server_addresses="127.0.0.1:8848",
credentials_provider=credentials_provider
)Install with Tessl CLI
npx tessl i tessl/pypi-nacos-sdk-python