or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

client-config.mdindex.mdv1-config.mdv1-naming.mdv2-config.mdv2-naming.md
tile.json

tessl/pypi-nacos-sdk-python

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

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/nacos-sdk-python@2.0.x

To install, run

npx @tessl/cli install tessl/pypi-nacos-sdk-python@2.0.0

index.mddocs/

Nacos SDK Python

A comprehensive Python SDK for the Nacos service discovery and configuration management platform. This SDK provides both synchronous (v1) and asynchronous (v2) APIs for dynamic service discovery, configuration management, and service registration with Nacos clusters.

Package Information

  • Package Name: nacos-sdk-python
  • Language: Python
  • Installation: pip install nacos-sdk-python
  • Minimum Python Version: Python 3.7+

Core Imports

Legacy V1 API (Synchronous)

from nacos import NacosClient, NacosException, DEFAULTS, DEFAULT_GROUP_NAME

Modern V2 API (Asynchronous)

from v2.nacos import (
    ClientConfig, ClientConfigBuilder,
    NacosConfigService, NacosNamingService,
    NacosException,
    ConfigParam,
    Instance, RegisterInstanceParam, DeregisterInstanceParam,
    ListInstanceParam, SubscribeServiceParam
)

Basic Usage

V1 API - Simple Configuration Management

from nacos import NacosClient

# Create client
client = NacosClient(
    server_addresses="127.0.0.1:8848",
    namespace="public",
    username="nacos",
    password="nacos"
)

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

# Publish configuration
client.publish_config("test-config", "DEFAULT_GROUP", "config content")

# Add configuration listener
def config_listener(args):
    print(f"Configuration changed: {args}")

client.add_config_watcher("test-config", "DEFAULT_GROUP", config_listener)

V2 API - Advanced Async Operations

import asyncio
from v2.nacos import ClientConfig, NacosConfigService, ConfigParam

async def main():
    # Create client configuration
    client_config = ClientConfig(
        server_addresses="127.0.0.1:8848",
        namespace_id="public",
        username="nacos",
        password="nacos"
    )
    
    # Create config service
    config_service = await NacosConfigService.create_config_service(client_config)
    
    # Get configuration
    param = ConfigParam(dataId="test-config", group="DEFAULT_GROUP")
    config = await config_service.get_config(param)
    print(config)
    
    # Publish configuration
    param.content = "new config content"
    await config_service.publish_config(param)
    
    # Shutdown
    await config_service.shutdown()

asyncio.run(main())

Architecture

Nacos SDK Python provides a dual-architecture approach:

  • V1 API (Legacy): Synchronous client with threading-based operations, suitable for simple use cases and backward compatibility
  • V2 API (Modern): Asynchronous client with GRPC support, advanced configuration options, and comprehensive data models

The V2 API introduces:

  • Configuration Service: Async operations for configuration management with encryption, filtering, and caching
  • Naming Service: Async operations for service discovery, registration, and health monitoring
  • Enhanced Models: Pydantic-based data models for type safety and validation
  • GRPC Support: High-performance communication protocol with connection pooling
  • Advanced Features: KMS encryption, TLS security, batch operations, and comprehensive error handling

Capabilities

V1 Configuration Management (Synchronous)

Legacy synchronous API for configuration operations including get, publish, remove, and watch operations with basic authentication and namespace support.

class NacosClient:
    def __init__(self, server_addresses=None, endpoint=None, namespace=None, 
                 ak=None, sk=None, username=None, password=None, **kwargs): ...
    def get_config(self, data_id: str, group: str, timeout=None, no_snapshot=None) -> str: ...
    def publish_config(self, data_id: str, group: str, content: str, 
                      app_name=None, config_type=None, timeout=None) -> bool: ...
    def remove_config(self, data_id: str, group: str, timeout=None) -> bool: ...
    def add_config_watcher(self, data_id: str, group: str, cb, content=None): ...
    def remove_config_watcher(self, data_id: str, group: str, cb, remove_all=False): ...

V1 Configuration Management

V1 Service Discovery (Synchronous)

Legacy synchronous API for service registration, discovery, health checks, and subscription operations with basic clustering support.

class NacosClient:
    def add_naming_instance(self, service_name: str, ip: str, port: int, 
                           cluster_name=None, weight=1.0, metadata=None, **kwargs) -> bool: ...
    def remove_naming_instance(self, service_name: str, ip: str, port: int, 
                              cluster_name=None, ephemeral=True, **kwargs) -> bool: ...
    def list_naming_instance(self, service_name: str, clusters=None, namespace_id=None, 
                            group_name=None, healthy_only=False) -> list: ...
    def send_heartbeat(self, service_name: str, ip: str, port: int, 
                      cluster_name=None, weight=1.0, metadata=None, ephemeral=True, **kwargs) -> dict: ...
    def subscribe(self, service_name: str, listener_name: str, listener_fn, 
                 cluster_names=None, **kwargs): ...
    def unsubscribe(self, service_name: str, listener_name=None): ...

V1 Service Discovery

V2 Configuration Management (Asynchronous)

Modern asynchronous API for configuration operations with encryption, filtering, caching, and advanced error handling.

class NacosConfigService:
    @staticmethod
    async def create_config_service(client_config: ClientConfig) -> 'NacosConfigService': ...
    async def get_config(self, param: ConfigParam) -> str: ...
    async def publish_config(self, param: ConfigParam) -> bool: ...
    async def remove_config(self, param: ConfigParam): ...
    async def add_listener(self, data_id: str, group: str, listener: Callable) -> None: ...
    async def remove_listener(self, data_id: str, group: str, listener: Callable): ...
    async def server_health(self) -> bool: ...
    async def shutdown(self): ...

V2 Configuration Management

V2 Service Discovery (Asynchronous)

Modern asynchronous API for service registration, discovery, batch operations, health monitoring, and subscription management with comprehensive data models.

class NacosNamingService:
    @staticmethod
    async def create_naming_service(client_config: ClientConfig) -> 'NacosNamingService': ...
    async def register_instance(self, request: RegisterInstanceParam) -> bool: ...
    async def batch_register_instances(self, request: BatchRegisterInstanceParam) -> bool: ...
    async def deregister_instance(self, request: DeregisterInstanceParam) -> bool: ...
    async def list_instances(self, request: ListInstanceParam) -> List[Instance]: ...
    async def get_service(self, request: GetServiceParam) -> Service: ...
    async def list_services(self, request: ListServiceParam) -> ServiceList: ...
    async def subscribe(self, request: SubscribeServiceParam) -> None: ...
    async def unsubscribe(self, request: SubscribeServiceParam) -> None: ...
    async def server_health(self) -> bool: ...
    async def shutdown(self) -> None: ...

V2 Service Discovery

Client Configuration

Configuration classes for setting up Nacos clients with support for authentication, security, networking, and operational parameters.

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): ...

class ClientConfigBuilder:
    def server_addresses(self, server_addresses: str) -> 'ClientConfigBuilder': ...
    def namespace_id(self, namespace_id: str) -> 'ClientConfigBuilder': ...
    def username(self, username: str) -> 'ClientConfigBuilder': ...
    def password(self, password: str) -> 'ClientConfigBuilder': ...
    def build(self) -> ClientConfig: ...

Client Configuration

Types

V2 Data Models

class Instance(BaseModel):
    instanceId: str = ''
    ip: str
    port: int
    weight: float = 1.0
    healthy: bool = True
    enabled: bool = True
    ephemeral: bool = True
    clusterName: str = ''
    serviceName: str = ''
    metadata: dict = {}

class ConfigParam(BaseModel):
    dataId: str
    group: str = 'DEFAULT_GROUP'
    content: str = ''
    type: str = ''
    appName: str = ''
    tag: str = ''
    md5: str = ''

class RegisterInstanceParam(BaseModel):
    ip: str
    port: int
    weight: float = 1.0
    enabled: bool = True
    healthy: bool = True
    metadata: Dict[str, str] = {}
    clusterName: str = ''
    serviceName: str
    groupName: str = 'DEFAULT_GROUP'
    ephemeral: bool = True

class Service(BaseModel):
    name: str
    groupName: str
    clusters: str
    cacheMillis: int
    hosts: List[Instance]
    lastRefTime: int
    checksum: str
    allIPs: bool
    reachProtectionThreshold: bool

class ServiceList(BaseModel):
    count: int
    services: List[str]

Exception Classes

class NacosException(Exception):
    def __init__(self, error_code=None, message="An error occurred"): ...

# V1 Exception (legacy)
class NacosRequestException(NacosException): ...

Constants

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"
}