Skip to content

Instances API

The InstancesAPI client provides methods for managing Incus instances (containers and virtual machines).

Usage

from incus_sdk import Client

async with Client() as client:
    # List all instances
    instances = await client.instances.list()

    # Create a new instance
    await client.instances.create(
        name="my-instance",
        source={
            "type": "image",
            "alias": "ubuntu/22.04"
        },
        wait=True
    )

    # Get an instance
    instance = await client.instances.get("my-instance")

    # Start an instance
    await client.instances.start("my-instance", wait=True)

    # Stop an instance
    await client.instances.stop("my-instance", wait=True)

    # Delete an instance
    await client.instances.delete("my-instance", wait=True)

Class Documentation

API client for Incus instances.

Source code in incus_sdk/api/instances.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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
class InstancesAPI:
    """API client for Incus instances."""

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

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

    async def list(self, recursion: int = 1) -> List[Instance]:
        """
        List all instances.

        Args:
            recursion: Level of recursion for the response.

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

        instances = []
        for instance_data in response.get("metadata", []):
            instances.append(Instance(client=self, **instance_data))

        return instances

    async def get(self, name: str) -> Instance:
        """
        Get an instance by name.

        Args:
            name: Name of the instance.

        Returns:
            Instance: The instance.
        """
        response = await self.client.get(
            f"/1.0/instances/{name}"
        )
        return Instance(client=self, **response.get("metadata", {}))

    async def create(
        self,
        name: str,
        source: Dict[str, Any],
        config: Dict[str, Any] = None,
        devices: Dict[str, Dict[str, Any]] = None,
        profiles: List[str] = None,
        ephemeral: bool = False,
        instance_type: str = "container",
        wait: bool = False,
    ) -> Union[Dict[str, Any], Instance]:
        """
        Create a new instance.

        Args:
            name: Name of the instance.
            source: Source configuration for the instance.
            config: Instance configuration.
            devices: Instance devices.
            profiles: List of profiles to apply.
            ephemeral: Whether the instance is ephemeral.
            instance_type: Type of instance (container or virtual-machine).
            wait: Whether to wait for the operation to complete.

        Returns:
            Union[Dict[str, Any], Instance]: The operation response or the created
            instance.
        """
        data = {"name": name, "source": source, "type": instance_type}

        if config:
            data["config"] = config
        if devices:
            data["devices"] = devices
        if profiles:
            data["profiles"] = profiles
        if ephemeral:
            data["ephemeral"] = ephemeral

        response = await self.client.post("/1.0/instances", data=data)

        if wait and "id" in response.get("metadata", {}):
            await self.client.wait_for_operation(response["metadata"]["id"])
            return await self.get(name)

        return response

    async def update(
        self, name: str, config: Dict[str, Any], wait: bool = False
    ) -> Dict[str, Any]:
        """
        Update an instance.

        Args:
            name: Name of the instance.
            config: New configuration.
            wait: Whether to wait for the operation to complete.

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

        if wait and "id" in response.get("metadata", {}):
            return await self.client.wait_for_operation(
                response["metadata"]["id"]
            )

        return response

    async def delete(self, name: str, wait: bool = False) -> Dict[str, Any]:
        """
        Delete an instance.

        Args:
            name: Name of the instance.
            wait: Whether to wait for the operation to complete.

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

        if wait and "id" in response.get("metadata", {}):
            return await self.client.wait_for_operation(
                response["metadata"]["id"]
            )

        return response

    async def _state_action(
        self,
        name: str,
        action: str,
        force: bool = False,
        stateful: bool = False,
        timeout: int = 30,
        wait: bool = False,
    ) -> Dict[str, Any]:
        """
        Perform a state action on an instance.

        Args:
            name: Name of the instance.
            action: Action to perform (start, stop, restart, freeze, unfreeze).
            force: Whether to force the action.
            stateful: Whether to preserve the instance state.
            timeout: Timeout in seconds.
            wait: Whether to wait for the operation to complete.

        Returns:
            Dict[str, Any]: The operation response.
        """
        data = {
            "action": action,
            "force": force,
            "stateful": stateful,
            "timeout": timeout,
        }

        response = await self.client.put(
            f"/1.0/instances/{name}/state", data=data
        )

        if wait and "id" in response.get("metadata", {}):
            return await self.client.wait_for_operation(
                response["metadata"]["id"]
            )

        return response

    async def start(
        self, name: str, stateful: bool = False, wait: bool = False
    ) -> Dict[str, Any]:
        """
        Start an instance.

        Args:
            name: Name of the instance.
            stateful: Whether to restore state.
            wait: Whether to wait for the operation to complete.

        Returns:
            Dict[str, Any]: The operation response.
        """
        return await self._state_action(name, "start", stateful=stateful, wait=wait)

    async def stop(
        self,
        name: str,
        force: bool = False,
        stateful: bool = False,
        wait: bool = False
    ) -> Dict[str, Any]:
        """
        Stop an instance.

        Args:
            name: Name of the instance.
            force: Whether to force stop.
            stateful: Whether to preserve state.
            wait: Whether to wait for the operation to complete.

        Returns:
            Dict[str, Any]: The operation response.
        """
        return await self._state_action(
            name, "stop", force=force, stateful=stateful, wait=wait
        )

    async def restart(
        self, name: str, force: bool = False, wait: bool = False
    ) -> Dict[str, Any]:
        """
        Restart an instance.

        Args:
            name: Name of the instance.
            force: Whether to force restart.
            wait: Whether to wait for the operation to complete.

        Returns:
            Dict[str, Any]: The operation response.
        """
        return await self._state_action(name, "restart", force=force, wait=wait)

    async def freeze(self, name: str, wait: bool = False) -> Dict[str, Any]:
        """
        Freeze an instance.

        Args:
            name: Name of the instance.
            wait: Whether to wait for the operation to complete.

        Returns:
            Dict[str, Any]: The operation response.
        """
        return await self._state_action(name, "freeze", wait=wait)

    async def unfreeze(self, name: str, wait: bool = False) -> Dict[str, Any]:
        """
        Unfreeze an instance.

        Args:
            name: Name of the instance.
            wait: Whether to wait for the operation to complete.

        Returns:
            Dict[str, Any]: The operation response.
        """
        return await self._state_action(name, "unfreeze", wait=wait)

    async def execute(
        self,
        name: str,
        command: List[str],
        environment: Dict[str, str] = None,
        wait_for_websocket: bool = False,
        record_output: bool = False,
        interactive: bool = False,
        width: int = None,
        height: int = None,
        user: int = None,
        group: int = None,
        cwd: str = None,
        wait: bool = False,
    ) -> Dict[str, Any]:
        """
        Execute a command in an instance.

        Args:
            name: Name of the instance.
            command: Command to execute.
            environment: Environment variables.
            wait_for_websocket: Whether to wait for a websocket connection.
            record_output: Whether to record the command output.
            interactive: Whether the command is interactive.
            width: Terminal width.
            height: Terminal height.
            user: User ID to run the command as.
            group: Group ID to run the command as.
            cwd: Working directory for the command.
            wait: Whether to wait for the operation to complete.

        Returns:
            Dict[str, Any]: The operation response.
        """
        data = {
            "command": command,
            "wait-for-websocket": wait_for_websocket,
            "record-output": record_output,
            "interactive": interactive,
        }

        if environment:
            data["environment"] = environment
        if width:
            data["width"] = width
        if height:
            data["height"] = height
        if user:
            data["user"] = user
        if group:
            data["group"] = group
        if cwd:
            data["cwd"] = cwd

        response = await self.client.post(
            f"/1.0/instances/{name}/exec", data=data
        )

        if wait and "id" in response.get("metadata", {}):
            return await self.client.wait_for_operation(
                response["metadata"]["id"]
            )

        return response

__init__(client)

Initialize a new InstancesAPI client.

Parameters:

Name Type Description Default
client APIClient

The base API client.

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

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

_state_action(name, action, force=False, stateful=False, timeout=30, wait=False) async

Perform a state action on an instance.

Parameters:

Name Type Description Default
name str

Name of the instance.

 required
action str

Action to perform (start, stop, restart, freeze, unfreeze).

 required
force bool

Whether to force the action.

 False
stateful bool

Whether to preserve the instance state.

 False
timeout int

Timeout in seconds.

 30
wait bool

Whether to wait for the operation to complete.

 False

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.
Source code in incus_sdk/api/instances.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
async def _state_action(
    self,
    name: str,
    action: str,
    force: bool = False,
    stateful: bool = False,
    timeout: int = 30,
    wait: bool = False,
) -> Dict[str, Any]:
    """
    Perform a state action on an instance.

    Args:
        name: Name of the instance.
        action: Action to perform (start, stop, restart, freeze, unfreeze).
        force: Whether to force the action.
        stateful: Whether to preserve the instance state.
        timeout: Timeout in seconds.
        wait: Whether to wait for the operation to complete.

    Returns:
        Dict[str, Any]: The operation response.
    """
    data = {
        "action": action,
        "force": force,
        "stateful": stateful,
        "timeout": timeout,
    }

    response = await self.client.put(
        f"/1.0/instances/{name}/state", data=data
    )

    if wait and "id" in response.get("metadata", {}):
        return await self.client.wait_for_operation(
            response["metadata"]["id"]
        )

    return response

create(name, source, config=None, devices=None, profiles=None, ephemeral=False, instance_type='container', wait=False) async

Create a new instance.

Parameters:

Name Type Description Default
name str

Name of the instance.

 required
source Dict[str, Any]

Source configuration for the instance.

 required
config Dict[str, Any]

Instance configuration.

 None
devices Dict[str, Dict[str, Any]]

Instance devices.

 None
profiles List[str]

List of profiles to apply.

 None
ephemeral bool

Whether the instance is ephemeral.

 False
instance_type str

Type of instance (container or virtual-machine).

 'container'
wait bool

Whether to wait for the operation to complete.

 False

Returns:

Type Description
Union[Dict[str, Any], Instance]

Union[Dict[str, Any], Instance]: The operation response or the created
Union[Dict[str, Any], Instance]

instance.
Source code in incus_sdk/api/instances.py 
 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
async def create(
    self,
    name: str,
    source: Dict[str, Any],
    config: Dict[str, Any] = None,
    devices: Dict[str, Dict[str, Any]] = None,
    profiles: List[str] = None,
    ephemeral: bool = False,
    instance_type: str = "container",
    wait: bool = False,
) -> Union[Dict[str, Any], Instance]:
    """
    Create a new instance.

    Args:
        name: Name of the instance.
        source: Source configuration for the instance.
        config: Instance configuration.
        devices: Instance devices.
        profiles: List of profiles to apply.
        ephemeral: Whether the instance is ephemeral.
        instance_type: Type of instance (container or virtual-machine).
        wait: Whether to wait for the operation to complete.

    Returns:
        Union[Dict[str, Any], Instance]: The operation response or the created
        instance.
    """
    data = {"name": name, "source": source, "type": instance_type}

    if config:
        data["config"] = config
    if devices:
        data["devices"] = devices
    if profiles:
        data["profiles"] = profiles
    if ephemeral:
        data["ephemeral"] = ephemeral

    response = await self.client.post("/1.0/instances", data=data)

    if wait and "id" in response.get("metadata", {}):
        await self.client.wait_for_operation(response["metadata"]["id"])
        return await self.get(name)

    return response

delete(name, wait=False) async

Delete an instance.

Parameters:

Name Type Description Default
name str

Name of the instance.

 required
wait bool

Whether to wait for the operation to complete.

 False

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.
Source code in incus_sdk/api/instances.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
async def delete(self, name: str, wait: bool = False) -> Dict[str, Any]:
    """
    Delete an instance.

    Args:
        name: Name of the instance.
        wait: Whether to wait for the operation to complete.

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

    if wait and "id" in response.get("metadata", {}):
        return await self.client.wait_for_operation(
            response["metadata"]["id"]
        )

    return response

execute(name, command, environment=None, wait_for_websocket=False, record_output=False, interactive=False, width=None, height=None, user=None, group=None, cwd=None, wait=False) async

Execute a command in an instance.

Parameters:

Name Type Description Default
name str

Name of the instance.

 required
command List[str]

Command to execute.

 required
environment Dict[str, str]

Environment variables.

 None
wait_for_websocket bool

Whether to wait for a websocket connection.

 False
record_output bool

Whether to record the command output.

 False
interactive bool

Whether the command is interactive.

 False
width int

Terminal width.

 None
height int

Terminal height.

 None
user int

User ID to run the command as.

 None
group int

Group ID to run the command as.

 None
cwd str

Working directory for the command.

 None
wait bool

Whether to wait for the operation to complete.

 False

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.
Source code in incus_sdk/api/instances.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
async def execute(
    self,
    name: str,
    command: List[str],
    environment: Dict[str, str] = None,
    wait_for_websocket: bool = False,
    record_output: bool = False,
    interactive: bool = False,
    width: int = None,
    height: int = None,
    user: int = None,
    group: int = None,
    cwd: str = None,
    wait: bool = False,
) -> Dict[str, Any]:
    """
    Execute a command in an instance.

    Args:
        name: Name of the instance.
        command: Command to execute.
        environment: Environment variables.
        wait_for_websocket: Whether to wait for a websocket connection.
        record_output: Whether to record the command output.
        interactive: Whether the command is interactive.
        width: Terminal width.
        height: Terminal height.
        user: User ID to run the command as.
        group: Group ID to run the command as.
        cwd: Working directory for the command.
        wait: Whether to wait for the operation to complete.

    Returns:
        Dict[str, Any]: The operation response.
    """
    data = {
        "command": command,
        "wait-for-websocket": wait_for_websocket,
        "record-output": record_output,
        "interactive": interactive,
    }

    if environment:
        data["environment"] = environment
    if width:
        data["width"] = width
    if height:
        data["height"] = height
    if user:
        data["user"] = user
    if group:
        data["group"] = group
    if cwd:
        data["cwd"] = cwd

    response = await self.client.post(
        f"/1.0/instances/{name}/exec", data=data
    )

    if wait and "id" in response.get("metadata", {}):
        return await self.client.wait_for_operation(
            response["metadata"]["id"]
        )

    return response

freeze(name, wait=False) async

Freeze an instance.

Parameters:

Name Type Description Default
name str

Name of the instance.

 required
wait bool

Whether to wait for the operation to complete.

 False

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.
Source code in incus_sdk/api/instances.py 
243
244
245
246
247
248
249
250
251
252
253
254
async def freeze(self, name: str, wait: bool = False) -> Dict[str, Any]:
    """
    Freeze an instance.

    Args:
        name: Name of the instance.
        wait: Whether to wait for the operation to complete.

    Returns:
        Dict[str, Any]: The operation response.
    """
    return await self._state_action(name, "freeze", wait=wait)

get(name) async

Get an instance by name.

Parameters:

Name Type Description Default
name str

Name of the instance.

 required

Returns:

Name Type Description
Instance Instance

The instance.
Source code in incus_sdk/api/instances.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
async def get(self, name: str) -> Instance:
    """
    Get an instance by name.

    Args:
        name: Name of the instance.

    Returns:
        Instance: The instance.
    """
    response = await self.client.get(
        f"/1.0/instances/{name}"
    )
    return Instance(client=self, **response.get("metadata", {}))

list(recursion=1) async

List all instances.

Parameters:

Name Type Description Default
recursion int

Level of recursion for the response.

 1

Returns:

Type Description
List[Instance]

List[Instance]: List of instances.
Source code in incus_sdk/api/instances.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
async def list(self, recursion: int = 1) -> List[Instance]:
    """
    List all instances.

    Args:
        recursion: Level of recursion for the response.

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

    instances = []
    for instance_data in response.get("metadata", []):
        instances.append(Instance(client=self, **instance_data))

    return instances

restart(name, force=False, wait=False) async

Restart an instance.

Parameters:

Name Type Description Default
name str

Name of the instance.

 required
force bool

Whether to force restart.

 False
wait bool

Whether to wait for the operation to complete.

 False

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.
Source code in incus_sdk/api/instances.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
async def restart(
    self, name: str, force: bool = False, wait: bool = False
) -> Dict[str, Any]:
    """
    Restart an instance.

    Args:
        name: Name of the instance.
        force: Whether to force restart.
        wait: Whether to wait for the operation to complete.

    Returns:
        Dict[str, Any]: The operation response.
    """
    return await self._state_action(name, "restart", force=force, wait=wait)

start(name, stateful=False, wait=False) async

Start an instance.

Parameters:

Name Type Description Default
name str

Name of the instance.

 required
stateful bool

Whether to restore state.

 False
wait bool

Whether to wait for the operation to complete.

 False

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.
Source code in incus_sdk/api/instances.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
async def start(
    self, name: str, stateful: bool = False, wait: bool = False
) -> Dict[str, Any]:
    """
    Start an instance.

    Args:
        name: Name of the instance.
        stateful: Whether to restore state.
        wait: Whether to wait for the operation to complete.

    Returns:
        Dict[str, Any]: The operation response.
    """
    return await self._state_action(name, "start", stateful=stateful, wait=wait)

stop(name, force=False, stateful=False, wait=False) async

Stop an instance.

Parameters:

Name Type Description Default
name str

Name of the instance.

 required
force bool

Whether to force stop.

 False
stateful bool

Whether to preserve state.

 False
wait bool

Whether to wait for the operation to complete.

 False

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.
Source code in incus_sdk/api/instances.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
async def stop(
    self,
    name: str,
    force: bool = False,
    stateful: bool = False,
    wait: bool = False
) -> Dict[str, Any]:
    """
    Stop an instance.

    Args:
        name: Name of the instance.
        force: Whether to force stop.
        stateful: Whether to preserve state.
        wait: Whether to wait for the operation to complete.

    Returns:
        Dict[str, Any]: The operation response.
    """
    return await self._state_action(
        name, "stop", force=force, stateful=stateful, wait=wait
    )

unfreeze(name, wait=False) async

Unfreeze an instance.

Parameters:

Name Type Description Default
name str

Name of the instance.

 required
wait bool

Whether to wait for the operation to complete.

 False

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.
Source code in incus_sdk/api/instances.py 
256
257
258
259
260
261
262
263
264
265
266
267
async def unfreeze(self, name: str, wait: bool = False) -> Dict[str, Any]:
    """
    Unfreeze an instance.

    Args:
        name: Name of the instance.
        wait: Whether to wait for the operation to complete.

    Returns:
        Dict[str, Any]: The operation response.
    """
    return await self._state_action(name, "unfreeze", wait=wait)

update(name, config, wait=False) async

Update an instance.

Parameters:

Name Type Description Default
name str

Name of the instance.

 required
config Dict[str, Any]

New configuration.

 required
wait bool

Whether to wait for the operation to complete.

 False

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.
Source code in incus_sdk/api/instances.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
async def update(
    self, name: str, config: Dict[str, Any], wait: bool = False
) -> Dict[str, Any]:
    """
    Update an instance.

    Args:
        name: Name of the instance.
        config: New configuration.
        wait: Whether to wait for the operation to complete.

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

    if wait and "id" in response.get("metadata", {}):
        return await self.client.wait_for_operation(
            response["metadata"]["id"]
        )

    return response

Methods

list

async def list(recursion: int = 1) -> List[Instance]

List all instances.

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

Returns: - A list of Instance objects.

get

async def get(name: str) -> Instance

Get an instance by name.

Parameters: - name: Name of the instance.

Returns: - An Instance object.

create

async def create(
    name: str,
    source: Dict[str, Any],
    config: Dict[str, Any] = None,
    devices: Dict[str, Dict[str, Any]] = None,
    profiles: List[str] = None,
    ephemeral: bool = False,
    instance_type: str = "container",
    wait: bool = False,
) -> Union[Dict[str, Any], Instance]

Create a new instance.

Parameters: - name: Name of the instance. - source: Source configuration for the instance. - config: Instance configuration (optional). - devices: Instance devices (optional). - profiles: List of profiles to apply (optional). - ephemeral: Whether the instance is ephemeral (default: False). - instance_type: Type of instance ("container" or "virtual-machine") (default: "container"). - wait: Whether to wait for the operation to complete (default: False).

Returns: - If wait is True, returns the created Instance object. - If wait is False, returns the operation response as a dictionary.

Example:

# Create a container from an image
container = await client.instances.create(
    name="my-container",
    source={
        "type": "image",
        "alias": "ubuntu/22.04"
    },
    config={
        "limits.cpu": "2",
        "limits.memory": "2GB"
    },
    devices={
        "root": {
            "path": "/",
            "pool": "default",
            "type": "disk"
        }
    },
    profiles=["default"],
    wait=True
)

# Create a virtual machine
vm = await client.instances.create(
    name="my-vm",
    source={
        "type": "image",
        "alias": "ubuntu/22.04"
    },
    instance_type="virtual-machine",
    wait=True
)

update

async def update(
    name: str, 
    config: Dict[str, Any], 
    wait: bool = False
) -> Dict[str, Any]

Update an instance.

Parameters: - name: Name of the instance. - config: New configuration. - wait: Whether to wait for the operation to complete (default: False).

Returns: - The operation response as a dictionary.

delete

async def delete(
    name: str, 
    wait: bool = False
) -> Dict[str, Any]

Delete an instance.

Parameters: - name: Name of the instance. - wait: Whether to wait for the operation to complete (default: False).

Returns: - The operation response as a dictionary.

start

async def start(
    name: str, 
    stateful: bool = False, 
    wait: bool = False
) -> Dict[str, Any]

Start an instance.

Parameters: - name: Name of the instance. - stateful: Whether to restore state (default: False). - wait: Whether to wait for the operation to complete (default: False).

Returns: - The operation response as a dictionary.

stop

async def stop(
    name: str, 
    force: bool = False, 
    stateful: bool = False, 
    timeout: int = 30, 
    wait: bool = False
) -> Dict[str, Any]

Stop an instance.

Parameters: - name: Name of the instance. - force: Whether to force the stop (default: False). - stateful: Whether to preserve the instance state (default: False). - timeout: Timeout in seconds (default: 30). - wait: Whether to wait for the operation to complete (default: False).

Returns: - The operation response as a dictionary.

restart

async def restart(
    name: str, 
    force: bool = False, 
    timeout: int = 30, 
    wait: bool = False
) -> Dict[str, Any]

Restart an instance.

Parameters: - name: Name of the instance. - force: Whether to force the restart (default: False). - timeout: Timeout in seconds (default: 30). - wait: Whether to wait for the operation to complete (default: False).

Returns: - The operation response as a dictionary.

freeze

async def freeze(
    name: str, 
    wait: bool = False
) -> Dict[str, Any]

Freeze an instance.

Parameters: - name: Name of the instance. - wait: Whether to wait for the operation to complete (default: False).

Returns: - The operation response as a dictionary.

unfreeze

async def unfreeze(
    name: str, 
    wait: bool = False
) -> Dict[str, Any]

Unfreeze an instance.

Parameters: - name: Name of the instance. - wait: Whether to wait for the operation to complete (default: False).

Returns: - The operation response as a dictionary.

execute

async def execute(
    name: str,
    command: List[str],
    environment: Dict[str, str] = None,
    wait_for_websocket: bool = False,
    record_output: bool = False,
    interactive: bool = False,
    width: int = None,
    height: int = None,
    user: int = None,
    group: int = None,
    cwd: str = None,
    wait: bool = False,
) -> Dict[str, Any]

Execute a command in an instance.

Parameters: - name: Name of the instance. - command: Command to execute as a list of strings. - environment: Environment variables (optional). - wait_for_websocket: Whether to wait for a websocket connection (default: False). - record_output: Whether to record the output (default: False). - interactive: Whether the command is interactive (default: False). - width: Terminal width (optional). - height: Terminal height (optional). - user: User ID to run the command as (optional). - group: Group ID to run the command as (optional). - cwd: Working directory (optional). - wait: Whether to wait for the operation to complete (default: False).

Returns: - The operation response as a dictionary.

Example:

# Execute a command and get the result
result = await client.instances.execute(
    name="my-container",
    command=["echo", "Hello, World!"],
    record_output=True,
    wait=True
)
print(result["metadata"]["output"]["stdout"])  # Prints: Hello, World!

get_state

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

Get the state of an instance.

Parameters: - name: Name of the instance.

Returns: - The instance state as a dictionary.

get_logs

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

Get the logs of an instance.

Parameters: - name: Name of the instance.

Returns: - The instance logs as a dictionary.

get_files

async def get_files(
    name: str, 
    path: str
) -> bytes

Get files from an instance.

Parameters: - name: Name of the instance. - path: Path to the file or directory.

Returns: - The file contents as bytes.

put_files

async def put_files(
    name: str, 
    path: str, 
    content: bytes
) -> Dict[str, Any]

Put files into an instance.

Parameters: - name: Name of the instance. - path: Path to the file or directory. - content: File contents as bytes.

Returns: - The operation response as a dictionary.