Skip to content

Cluster API

The ClusterAPI client provides methods for managing Incus clusters.

Usage

from incus_sdk import Client

async with Client() as client:
    # Get cluster information
    cluster = await client.cluster.get()

    # List all cluster members
    members = await client.cluster.list_members()

    # Get a cluster member by name
    member = await client.cluster.get_member("node1")

    # Add a new member to the cluster
    await client.cluster.add_member(
        name="node2",
        url="10.0.0.2:8443",
        config={
            "enabled": True
        }
    )

    # Update a cluster member
    await client.cluster.update_member(
        "node2",
        {
            "config": {
                "enabled": False
            }
        }
    )

    # Delete a cluster member
    await client.cluster.delete_member("node2")

Class Documentation

API client for Incus cluster.

Source code in incus_sdk/api/cluster.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
class ClusterAPI:
    """API client for Incus cluster."""

    def __init__(self, client: APIClient):
        """
        Initialize a new ClusterAPI client.

        Args:
            client: The base API client.
        """
        self.client = client

    async def get(self) -> Cluster:
        """
        Get cluster information.

        Returns:
            Cluster: The cluster.
        """
        response = await self.client.get("/1.0/cluster")
        return Cluster(client=self, **response.get("metadata", {}))

    async def update(self, config: Dict[str, Any]) -> Dict[str, Any]:
        """
        Update cluster configuration.

        Args:
            config: New configuration.

        Returns:
            Dict[str, Any]: The operation response.
        """
        return await self.client.put("/1.0/cluster", data=config)

    async def list_members(self, recursion: int = 1) -> List[ClusterMember]:
        """
        List all cluster members.

        Args:
            recursion: Level of recursion for the response.

        Returns:
            List[ClusterMember]: List of cluster members.
        """
        params = {"recursion": recursion}
        response = await self.client.get("/1.0/cluster/members", params=params)

        members = []
        for member_data in response.get("metadata", []):
            members.append(ClusterMember(client=self, **member_data))

        return members

    async def get_member(self, name: str) -> ClusterMember:
        """
        Get a cluster member by name.

        Args:
            name: Name of the cluster member.

        Returns:
            ClusterMember: The cluster member.
        """
        response = await self.client.get(f"/1.0/cluster/members/{name}")
        return ClusterMember(client=self, **response.get("metadata", {}))

    async def add_member(
        self, name: str, url: str, config: Dict[str, Any] = None
    ) -> Dict[str, Any]:
        """
        Add a new member to the cluster.

        Args:
            name: Name of the cluster member.
            url: URL of the cluster member.
            config: Member configuration.

        Returns:
            Dict[str, Any]: The operation response.
        """
        data = {"server_name": name, "server_address": url}

        if config:
            data["config"] = config

        return await self.client.post("/1.0/cluster/members", data=data)

    async def update_member(self, name: str, config: Dict[str, Any]) -> Dict[str, Any]:
        """
        Update a cluster member.

        Args:
            name: Name of the cluster member.
            config: New configuration.

        Returns:
            Dict[str, Any]: The operation response.
        """
        return await self.client.patch(f"/1.0/cluster/members/{name}", data=config)

    async def delete_member(self, name: str) -> Dict[str, Any]:
        """
        Delete a cluster member.

        Args:
            name: Name of the cluster member.

        Returns:
            Dict[str, Any]: The operation response.
        """
        return await self.client.delete(f"/1.0/cluster/members/{name}")

    async def rename_member(self, name: str, new_name: str) -> Dict[str, Any]:
        """
        Rename a cluster member.

        Args:
            name: Current name of the cluster member.
            new_name: New name for the cluster member.

        Returns:
            Dict[str, Any]: The operation response.
        """
        data = {"server_name": new_name}

        return await self.client.post(f"/1.0/cluster/members/{name}", data=data)

__init__(client)

Initialize a new ClusterAPI client.

Parameters:

Name Type Description Default
client APIClient

The base API client.

 required
Source code in incus_sdk/api/cluster.py 
14
15
16
17
18
19
20
21
def __init__(self, client: APIClient):
    """
    Initialize a new ClusterAPI client.

    Args:
        client: The base API client.
    """
    self.client = client

add_member(name, url, config=None) async

Add a new member to the cluster.

Parameters:

Name Type Description Default
name str

Name of the cluster member.

 required
url str

URL of the cluster member.

 required
config Dict[str, Any]

Member configuration.

 None

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.
Source code in incus_sdk/api/cluster.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
async def add_member(
    self, name: str, url: str, config: Dict[str, Any] = None
) -> Dict[str, Any]:
    """
    Add a new member to the cluster.

    Args:
        name: Name of the cluster member.
        url: URL of the cluster member.
        config: Member configuration.

    Returns:
        Dict[str, Any]: The operation response.
    """
    data = {"server_name": name, "server_address": url}

    if config:
        data["config"] = config

    return await self.client.post("/1.0/cluster/members", data=data)

delete_member(name) async

Delete a cluster member.

Parameters:

Name Type Description Default
name str

Name of the cluster member.

 required

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.
Source code in incus_sdk/api/cluster.py 
111
112
113
114
115
116
117
118
119
120
121
async def delete_member(self, name: str) -> Dict[str, Any]:
    """
    Delete a cluster member.

    Args:
        name: Name of the cluster member.

    Returns:
        Dict[str, Any]: The operation response.
    """
    return await self.client.delete(f"/1.0/cluster/members/{name}")

get() async

Get cluster information.

Returns:

Name Type Description
Cluster Cluster

The cluster.
Source code in incus_sdk/api/cluster.py 
23
24
25
26
27
28
29
30
31
async def get(self) -> Cluster:
    """
    Get cluster information.

    Returns:
        Cluster: The cluster.
    """
    response = await self.client.get("/1.0/cluster")
    return Cluster(client=self, **response.get("metadata", {}))

get_member(name) async

Get a cluster member by name.

Parameters:

Name Type Description Default
name str

Name of the cluster member.

 required

Returns:

Name Type Description
ClusterMember ClusterMember

The cluster member.
Source code in incus_sdk/api/cluster.py 
64
65
66
67
68
69
70
71
72
73
74
75
async def get_member(self, name: str) -> ClusterMember:
    """
    Get a cluster member by name.

    Args:
        name: Name of the cluster member.

    Returns:
        ClusterMember: The cluster member.
    """
    response = await self.client.get(f"/1.0/cluster/members/{name}")
    return ClusterMember(client=self, **response.get("metadata", {}))

list_members(recursion=1) async

List all cluster members.

Parameters:

Name Type Description Default
recursion int

Level of recursion for the response.

 1

Returns:

Type Description
List[ClusterMember]

List[ClusterMember]: List of cluster members.
Source code in incus_sdk/api/cluster.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
async def list_members(self, recursion: int = 1) -> List[ClusterMember]:
    """
    List all cluster members.

    Args:
        recursion: Level of recursion for the response.

    Returns:
        List[ClusterMember]: List of cluster members.
    """
    params = {"recursion": recursion}
    response = await self.client.get("/1.0/cluster/members", params=params)

    members = []
    for member_data in response.get("metadata", []):
        members.append(ClusterMember(client=self, **member_data))

    return members

rename_member(name, new_name) async

Rename a cluster member.

Parameters:

Name Type Description Default
name str

Current name of the cluster member.

 required
new_name str

New name for the cluster member.

 required

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.
Source code in incus_sdk/api/cluster.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
async def rename_member(self, name: str, new_name: str) -> Dict[str, Any]:
    """
    Rename a cluster member.

    Args:
        name: Current name of the cluster member.
        new_name: New name for the cluster member.

    Returns:
        Dict[str, Any]: The operation response.
    """
    data = {"server_name": new_name}

    return await self.client.post(f"/1.0/cluster/members/{name}", data=data)

update(config) async

Update cluster configuration.

Parameters:

Name Type Description Default
config Dict[str, Any]

New configuration.

 required

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.
Source code in incus_sdk/api/cluster.py 
33
34
35
36
37
38
39
40
41
42
43
async def update(self, config: Dict[str, Any]) -> Dict[str, Any]:
    """
    Update cluster configuration.

    Args:
        config: New configuration.

    Returns:
        Dict[str, Any]: The operation response.
    """
    return await self.client.put("/1.0/cluster", data=config)

update_member(name, config) async

Update a cluster member.

Parameters:

Name Type Description Default
name str

Name of the cluster member.

 required
config Dict[str, Any]

New configuration.

 required

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.
Source code in incus_sdk/api/cluster.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
async def update_member(self, name: str, config: Dict[str, Any]) -> Dict[str, Any]:
    """
    Update a cluster member.

    Args:
        name: Name of the cluster member.
        config: New configuration.

    Returns:
        Dict[str, Any]: The operation response.
    """
    return await self.client.patch(f"/1.0/cluster/members/{name}", data=config)

Methods

get

async def get() -> Cluster

Get cluster information.

Returns: - A Cluster object.

update

async def update(
    config: Dict[str, Any]
) -> Dict[str, Any]

Update cluster configuration.

Parameters: - config: New configuration as a dictionary.

Returns: - The operation response as a dictionary.

list_members

async def list_members(
    recursion: int = 1
) -> List[ClusterMember]

List all cluster members.

Parameters: - recursion: Level of recursion for the response (default: 1).

Returns: - A list of ClusterMember objects.

get_member

async def get_member(
    name: str
) -> ClusterMember

Get a cluster member by name.

Parameters: - name: Name of the cluster member.

Returns: - A ClusterMember object.

add_member

async def add_member(
    name: str, 
    url: str, 
    config: Dict[str, Any] = None
) -> Dict[str, Any]

Add a new member to the cluster.

Parameters: - name: Name of the cluster member. - url: URL of the cluster member. - config: Member configuration as a dictionary (optional).

Returns: - The operation response as a dictionary.

Example:

# Add a new member to the cluster
await client.cluster.add_member(
    name="node2",
    url="10.0.0.2:8443",
    config={
        "enabled": True,
        "role": "worker"
    }
)

update_member

async def update_member(
    name: str, 
    config: Dict[str, Any]
) -> Dict[str, Any]

Update a cluster member.

Parameters: - name: Name of the cluster member. - config: New configuration as a dictionary.

Returns: - The operation response as a dictionary.

delete_member

async def delete_member(
    name: str
) -> Dict[str, Any]

Delete a cluster member.

Parameters: - name: Name of the cluster member.

Returns: - The operation response as a dictionary.

rename_member

async def rename_member(
    name: str, 
    new_name: str
) -> Dict[str, Any]

Rename a cluster member.

Parameters: - name: Current name of the cluster member. - new_name: New name for the cluster member.

Returns: - The operation response as a dictionary.