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 Service Discovery (Synchronous)

Name: tessl/pypi-nacos-sdk-python
Author: tessl

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

Capabilities

Service Instance Registration

def add_naming_instance(self, service_name: str, ip: str, port: int, cluster_name=None, 
                       weight=1.0, metadata=None, ephemeral=True, enabled=True, 
                       healthy=True, group_name=None) -> bool:
    """
    Register a service instance.
    
    Args:
        service_name (str): Name of the service
        ip (str): IP address of the instance
        port (int): Port number of the instance
        cluster_name (str, optional): Cluster name
        weight (float): Instance weight for load balancing (default: 1.0)
        metadata (dict, optional): Instance metadata
        ephemeral (bool): Whether instance is ephemeral (default: True)
        enabled (bool): Whether instance is enabled (default: True)
        healthy (bool): Whether instance is healthy (default: True)
        group_name (str, optional): Service group name
        
    Returns:
        bool: True if registration successful
        
    Raises:
        NacosException: If parameters are invalid
        NacosRequestException: If request fails
    """

Usage example:

# Basic instance registration
client.add_naming_instance("user-service", "192.168.1.100", 8080)

# Registration with metadata
metadata = {
    "version": "1.0.0",
    "region": "us-west-1",
    "zone": "us-west-1a"
}
client.add_naming_instance(
    service_name="user-service",
    ip="192.168.1.100", 
    port=8080,
    cluster_name="production",
    weight=1.5,
    metadata=metadata
)

# Registration with custom group
client.add_naming_instance(
    service_name="user-service",
    ip="192.168.1.100",
    port=8080,
    group_name="microservices"
)

Service Instance Deregistration

Remove service instances from Nacos registry.

def remove_naming_instance(self, service_name: str, ip: str, port: int, cluster_name=None, 
                          ephemeral=True, group_name=None) -> bool:
    """
    Deregister a service instance.
    
    Args:
        service_name (str): Name of the service
        ip (str): IP address of the instance
        port (int): Port number of the instance
        cluster_name (str, optional): Cluster name
        ephemeral (bool): Whether instance is ephemeral (default: True)
        group_name (str, optional): Service group name
        
    Returns:
        bool: True if deregistration successful
        
    Raises:
        NacosException: If parameters are invalid
        NacosRequestException: If request fails
    """

Usage example:

# Basic instance deregistration
client.remove_naming_instance("user-service", "192.168.1.100", 8080)

# Deregistration from specific cluster
client.remove_naming_instance(
    service_name="user-service",
    ip="192.168.1.100",
    port=8080,
    cluster_name="production"
)

# Deregistration from custom group
client.remove_naming_instance(
    service_name="user-service",
    ip="192.168.1.100",
    port=8080,
    group_name="microservices"
)

Service Instance Modification

Update existing service instance properties.

def modify_naming_instance(self, service_name: str, ip: str, port: int, cluster_name=None, 
                          weight=None, metadata=None, enabled=None, group_name=None) -> bool:
    """
    Modify a service instance.
    
    Args:
        service_name (str): Name of the service
        ip (str): IP address of the instance
        port (int): Port number of the instance
        cluster_name (str, optional): Cluster name
        weight (float, optional): New instance weight
        metadata (dict, optional): New instance metadata
        enabled (bool, optional): New enabled status
        group_name (str, optional): Service group name
        
    Returns:
        bool: True if modification successful
        
    Raises:
        NacosException: If parameters are invalid
        NacosRequestException: If request fails
    """

Usage example:

# Update instance weight
client.modify_naming_instance("user-service", "192.168.1.100", 8080, weight=2.0)

# Update metadata
new_metadata = {
    "version": "1.1.0",
    "region": "us-west-1",
    "status": "updated"
}
client.modify_naming_instance(
    service_name="user-service",
    ip="192.168.1.100",
    port=8080,
    metadata=new_metadata
)

# Disable instance
client.modify_naming_instance("user-service", "192.168.1.100", 8080, enabled=False)

Service Instance Listing

Retrieve lists of service instances.

def list_naming_instance(self, service_name: str, clusters=None, namespace_id=None, 
                        group_name=None, healthy_only=False) -> list:
    """
    List service instances.
    
    Args:
        service_name (str): Name of the service
        clusters (list, optional): List of cluster names to filter
        namespace_id (str, optional): Namespace ID
        group_name (str, optional): Service group name
        healthy_only (bool): Return only healthy instances (default: False)
        
    Returns:
        list: List of service instances
        
    Raises:
        NacosException: If parameters are invalid
        NacosRequestException: If request fails
    """

Usage example:

# List all instances
instances = client.list_naming_instance("user-service")
for instance in instances:
    print(f"Instance: {instance['ip']}:{instance['port']}")

# List healthy instances only
healthy_instances = client.list_naming_instance("user-service", healthy_only=True)

# List instances from specific clusters
instances = client.list_naming_instance(
    service_name="user-service",
    clusters=["production", "staging"]
)

# List instances from custom group and namespace
instances = client.list_naming_instance(
    service_name="user-service",
    namespace_id="production",
    group_name="microservices"
)

Service Instance Querying

Get detailed information about a specific service instance.

def get_naming_instance(self, service_name: str, ip: str, port: int, cluster_name=None) -> dict:
    """
    Get detailed information about a specific service instance.
    
    Args:
        service_name (str): Name of the service
        ip (str): IP address of the instance
        port (int): Port number of the instance
        cluster_name (str, optional): Cluster name
        
    Returns:
        dict: Instance details including metadata, health status, and configuration
        
    Raises:
        NacosException: If parameters are invalid
        NacosRequestException: If request fails
    """

Usage example:

# Get instance details
instance = client.get_naming_instance("user-service", "192.168.1.100", 8080)
print(f"Instance details: {instance}")
print(f"Health status: {instance.get('healthy', 'unknown')}")
print(f"Metadata: {instance.get('metadata', {})}")

# Get instance from specific cluster
instance = client.get_naming_instance(
    service_name="user-service",
    ip="192.168.1.100",
    port=8080,
    cluster_name="production"
)

Health Check and Heartbeat

Send heartbeat signals to maintain instance health status.

def send_heartbeat(self, service_name: str, ip: str, port: int, cluster_name=None, 
                  weight=1.0, metadata=None, ephemeral=True, group_name=None) -> dict:
    """
    Send heartbeat for a service instance.
    
    Args:
        service_name (str): Name of the service
        ip (str): IP address of the instance
        port (int): Port number of the instance
        cluster_name (str, optional): Cluster name
        weight (float): Instance weight (default: 1.0)
        metadata (dict, optional): Instance metadata
        ephemeral (bool): Whether instance is ephemeral (default: True)
        group_name (str, optional): Service group name
        
    Returns:
        dict: Heartbeat response containing server time and other info
        
    Raises:
        NacosException: If parameters are invalid
        NacosRequestException: If request fails
    """

Usage example:

# Send basic heartbeat
response = client.send_heartbeat("user-service", "192.168.1.100", 8080)
print(f"Server time: {response.get('clientBeatInterval', 'unknown')}")

# Send heartbeat with metadata
metadata = {"status": "running", "cpu": "20%", "memory": "512MB"}
response = client.send_heartbeat(
    service_name="user-service",
    ip="192.168.1.100",
    port=8080,
    weight=1.5,
    metadata=metadata
)

# Periodic heartbeat example
import time
import threading

def heartbeat_worker():
    while True:
        try:
            client.send_heartbeat("user-service", "192.168.1.100", 8080)
            time.sleep(5)  # Send heartbeat every 5 seconds
        except Exception as e:
            print(f"Heartbeat failed: {e}")

heartbeat_thread = threading.Thread(target=heartbeat_worker, daemon=True)
heartbeat_thread.start()

Service Subscription

Subscribe to service change notifications.

def subscribe(self, service_name: str, listener_fn, cluster_name=None, 
             group_name=None):
    """
    Subscribe to service changes.
    
    Args:
        service_name (str): Name of the service to monitor
        listener_fn (callable): Callback function for service changes
        cluster_name (str, optional): Cluster name to monitor
        group_name (str, optional): Service group name
    """

def unsubscribe(self, service_name: str, listener_fn=None):
    """
    Unsubscribe from service changes.
    
    Args:
        service_name (str): Name of the service
        listener_fn (callable, optional): Listener function to remove (if None, removes all)
    """

def stop_subscribe(self):
    """Stop all service subscriptions."""

Usage example:

def service_change_handler(event):
    print(f"Service changed: {event}")
    # Handle service instance changes
    instances = event.get('hosts', [])
    for instance in instances:
        print(f"  Instance: {instance['ip']}:{instance['port']} - {instance['healthy']}")

# Subscribe to service changes
client.subscribe(
    service_name="user-service",
    listener_fn=service_change_handler
)

# Subscribe to specific cluster
client.subscribe(
    service_name="user-service",
    listener_fn=service_change_handler,
    cluster_name="production"
)

# Subscribe with custom group
client.subscribe(
    service_name="user-service",
    listener_fn=service_change_handler,
    group_name="microservices"
)

# Unsubscribe specific listener
client.unsubscribe("user-service", service_change_handler)

# Unsubscribe all listeners for a service
client.unsubscribe("user-service")

# Stop all subscriptions
client.stop_subscribe()

Data Structures

Service instance responses typically contain:

# Example instance data structure
{
    "instanceId": "192.168.1.100#8080#DEFAULT#user-service",
    "ip": "192.168.1.100",
    "port": 8080,
    "weight": 1.0,
    "healthy": True,
    "enabled": True,
    "ephemeral": True,
    "clusterName": "DEFAULT",
    "serviceName": "user-service",
    "metadata": {
        "version": "1.0.0",
        "region": "us-west-1"
    }
}

# Example service change event structure
{
    "name": "user-service",
    "groupName": "DEFAULT_GROUP",
    "clusters": "DEFAULT",
    "cacheMillis": 10000,
    "hosts": [
        {
            "instanceId": "192.168.1.100#8080#DEFAULT#user-service",
            "ip": "192.168.1.100",
            "port": 8080,
            "weight": 1.0,
            "healthy": True,
            "enabled": True,
            "ephemeral": True,
            "clusterName": "DEFAULT",
            "serviceName": "user-service",
            "metadata": {}
        }
    ],
    "lastRefTime": 1640995200000,
    "checksum": "abc123",
    "allIPs": False,
    "reachProtectionThreshold": False
}

Install with Tessl CLI