Node Management

def inspect(x: Union[str, List[str]]) -> Union[Node, List[Node]]:
    """
    Inspect one or more nodes in the swarm.
    
    Parameters:
    - x: Node name(s) or ID(s)
    
    Returns:
    Node object(s) with detailed information
    """

def list() -> List[Node]:
    """
    List all nodes in the swarm.
    
    Returns:
    List of Node objects
    """

def promote(x: Union[str, List[str]]) -> None:
    """
    Promote worker nodes to manager nodes.
    
    Parameters:
    - x: Node name(s) or ID(s) to promote
    """

def demote(x: Union[str, List[str]]) -> None:
    """
    Demote manager nodes to worker nodes.
    
    Parameters:
    - x: Node name(s) or ID(s) to demote
    """

def update(
    node: Union[str, Node],
    availability: Optional[str] = None,
    labels_add: Optional[Dict[str, str]] = None,
    rm_labels: Optional[List[str]] = None,
    role: Optional[str] = None
) -> None:
    """
    Update node configuration.
    
    Parameters:
    - node: Node name, ID, or Node object
    - availability: Node availability (active, pause, drain)
    - labels_add: Labels to add to the node
    - rm_labels: Label keys to remove from the node
    - role: Node role (manager, worker)
    """

def ps(x: Optional[Union[str, List[str]]] = None) -> List[Task]:
    """
    List tasks running on nodes.
    
    Parameters:
    - x: Node name(s) or ID(s) (current node if none specified)
    
    Returns:
    List of Task objects
    """

def remove(
    x: Union[str, List[str]], 
    force: bool = False
) -> None:
    """
    Remove nodes from the swarm.
    
    Parameters:
    - x: Node name(s) or ID(s) to remove
    - force: Force removal of nodes
    """

from python_on_whales import docker

# List all nodes in swarm
nodes = docker.node.list()
for node in nodes:
    print(f"Node: {node.description.hostname} - Role: {node.spec.role}")

# Promote worker to manager
docker.node.promote("worker-01")

# Update node availability and add labels
docker.node.update(
    "node-01",
    availability="drain",
    labels_add={"maintenance": "true", "zone": "us-east-1a"}
)

# List tasks on specific node
tasks = docker.node.ps("manager-01")
for task in tasks:
    print(f"Task: {task.name} - Service: {task.service_id}")

class Node:
    id: str
    version: DockerObjectVersion
    created_at: datetime
    updated_at: datetime
    spec: NodeSpec
    description: NodeDescription
    status: NodeStatus
    manager_status: Optional[ManagerStatus]
    
    def update(
        self,
        availability: Optional[str] = None,
        labels_add: Optional[Dict[str, str]] = None,
        rm_labels: Optional[List[str]] = None,
        role: Optional[str] = None
    ) -> None:
        """Update this node's configuration."""
    
    def ps(self) -> List[Task]:
        """List tasks running on this node."""

class NodeSpec:
    name: Optional[str]
    labels: Dict[str, str]
    role: str  # "manager" or "worker"
    availability: str  # "active", "pause", "drain"

class NodeDescription:
    hostname: str
    platform: Dict[str, str]
    resources: Dict[str, Any]
    engine: Dict[str, Any]

class NodeStatus:
    state: str  # "ready", "down", "unknown"
    message: Optional[str]
    addr: str

class ManagerStatus:
    leader: bool
    reachability: str  # "reachable", "unreachable"
    addr: str

class DockerObjectVersion:
    index: int