Skip to content

Storage Pools API

The StoragePoolsAPI client provides methods for managing Incus storage pools and volumes.

Usage

from incus_sdk import Client

async with Client() as client:
    # List all storage pools
    pools = await client.storage_pools.list()

    # Get a storage pool by name
    pool = await client.storage_pools.get("default")

    # Create a new storage pool
    await client.storage_pools.create(
        name="my-pool",
        driver="dir",
        config={
            "source": "/var/lib/incus/storage-pools/my-pool"
        },
        description="My custom storage pool"
    )

    # List volumes in a storage pool
    volumes = await client.storage_pools.list_volumes("default")

    # Create a new volume
    await client.storage_pools.create_volume(
        pool_name="default",
        volume_name="my-volume",
        volume_type="custom",
        config={
            "size": "10GB"
        }
    )

    # Delete a storage pool
    await client.storage_pools.delete("my-pool")

Class Documentation

API client for Incus storage pools.

Source code in incus_sdk/api/storage_pools.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
class StoragePoolsAPI:
    """API client for Incus storage pools."""

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

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

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

        Args:
            recursion: Level of recursion for the response.

        Returns:
            List[StoragePool]: List of storage pools.
        """
        params = {"recursion": recursion}
        response = await self.client.get("/1.0/storage-pools", params=params)

        pools = []
        for pool_data in response.get("metadata", []):
            pools.append(StoragePool(client=self, **pool_data))

        return pools

    async def get(self, name: str) -> StoragePool:
        """
        Get a storage pool by name.

        Args:
            name: Name of the storage pool.

        Returns:
            StoragePool: The storage pool.
        """
        response = await self.client.get(f"/1.0/storage-pools/{name}")
        return StoragePool(client=self, **response.get("metadata", {}))

    async def create(
        self,
        name: str,
        driver: str,
        config: Dict[str, Any] = None,
        description: str = None,
    ) -> Dict[str, Any]:
        """
        Create a new storage pool.

        Args:
            name: Name of the storage pool.
            driver: Storage driver.
            config: Storage pool configuration.
            description: Description of the storage pool.

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

        if config:
            data["config"] = config
        if description:
            data["description"] = description

        return await self.client.post("/1.0/storage-pools", data=data)

    async def update(self, name: str, config: Dict[str, Any]) -> Dict[str, Any]:
        """
        Update a storage pool.

        Args:
            name: Name of the storage pool.
            config: New configuration.

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

    async def replace(self, name: str, config: Dict[str, Any]) -> Dict[str, Any]:
        """
        Replace a storage pool configuration.

        Args:
            name: Name of the storage pool.
            config: New configuration.

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

    async def delete(self, name: str) -> Dict[str, Any]:
        """
        Delete a storage pool.

        Args:
            name: Name of the storage pool.

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

    async def list_volumes(
        self, pool_name: str, volume_type: str = None, recursion: int = 1
    ) -> List[StorageVolume]:
        """
        List volumes in a storage pool.

        Args:
            pool_name: Name of the storage pool.
            volume_type: Type of volumes to list.
            recursion: Level of recursion for the response.

        Returns:
            List[StorageVolume]: List of storage volumes.
        """
        path = f"/1.0/storage-pools/{pool_name}/volumes"
        if volume_type:
            path = f"{path}/{volume_type}"

        params = {"recursion": recursion}
        response = await self.client.get(path, params=params)

        volumes = []
        for volume_data in response.get("metadata", []):
            volume_data["pool"] = pool_name
            volumes.append(StorageVolume(client=self, **volume_data))

        return volumes

    async def get_volume(
        self, pool_name: str, volume_name: str, volume_type: str
    ) -> StorageVolume:
        """
        Get a storage volume by name.

        Args:
            pool_name: Name of the storage pool.
            volume_name: Name of the volume.
            volume_type: Type of volume.

        Returns:
            StorageVolume: The storage volume.
        """
        response = await self.client.get(
            f"/1.0/storage-pools/{pool_name}/volumes/{volume_type}/{volume_name}"
        )
        volume_data = response.get("metadata", {})
        volume_data["pool"] = pool_name
        return StorageVolume(client=self, **volume_data)

    async def create_volume(
        self,
        pool_name: str,
        volume_name: str,
        volume_type: str,
        config: Dict[str, Any] = None,
    ) -> Dict[str, Any]:
        """
        Create a new storage volume.

        Args:
            pool_name: Name of the storage pool.
            volume_name: Name of the volume.
            volume_type: Type of volume.
            config: Volume configuration.

        Returns:
            Dict[str, Any]: The operation response.
        """
        data = {"name": volume_name, "type": volume_type}

        if config:
            data["config"] = config

        return await self.client.post(
            f"/1.0/storage-pools/{pool_name}/volumes", data=data
        )

    async def update_volume(
        self, pool_name: str, volume_name: str, volume_type: str, config: Dict[str, Any]
    ) -> Dict[str, Any]:
        """
        Update a storage volume.

        Args:
            pool_name: Name of the storage pool.
            volume_name: Name of the volume.
            volume_type: Type of volume.
            config: New configuration.

        Returns:
            Dict[str, Any]: The operation response.
        """
        return await self.client.patch(
            f"/1.0/storage-pools/{pool_name}/volumes/{volume_type}/{volume_name}",
            data=config,
        )

    async def replace_volume(
        self, pool_name: str, volume_name: str, volume_type: str, config: Dict[str, Any]
    ) -> Dict[str, Any]:
        """
        Replace a storage volume configuration.

        Args:
            pool_name: Name of the storage pool.
            volume_name: Name of the volume.
            volume_type: Type of volume.
            config: New configuration.

        Returns:
            Dict[str, Any]: The operation response.
        """
        return await self.client.put(
            f"/1.0/storage-pools/{pool_name}/volumes/{volume_type}/{volume_name}",
            data=config,
        )

    async def delete_volume(
        self, pool_name: str, volume_name: str, volume_type: str
    ) -> Dict[str, Any]:
        """
        Delete a storage volume.

        Args:
            pool_name: Name of the storage pool.
            volume_name: Name of the volume.
            volume_type: Type of volume.

        Returns:
            Dict[str, Any]: The operation response.
        """
        return await self.client.delete(
            f"/1.0/storage-pools/{pool_name}/volumes/{volume_type}/{volume_name}"
        )

    async def rename_volume(
        self, pool_name: str, volume_name: str, volume_type: str, new_name: str
    ) -> Dict[str, Any]:
        """
        Rename a storage volume.

        Args:
            pool_name: Name of the storage pool.
            volume_name: Current name of the volume.
            volume_type: Type of volume.
            new_name: New name for the volume.

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

        return await self.client.post(
            f"/1.0/storage-pools/{pool_name}/volumes/{volume_type}/{volume_name}",
            data=data,
        )

__init__(client)

Initialize a new StoragePoolsAPI client.

Parameters:

Name Type Description Default
client APIClient

The base API client.

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

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

create(name, driver, config=None, description=None) async

Create a new storage pool.

Parameters:

Name Type Description Default
name str

Name of the storage pool.

required
driver str

Storage driver.

required
config Dict[str, Any]

Storage pool configuration.

None
description str

Description of the storage pool.

None

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.

Source code in incus_sdk/api/storage_pools.py
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
async def create(
    self,
    name: str,
    driver: str,
    config: Dict[str, Any] = None,
    description: str = None,
) -> Dict[str, Any]:
    """
    Create a new storage pool.

    Args:
        name: Name of the storage pool.
        driver: Storage driver.
        config: Storage pool configuration.
        description: Description of the storage pool.

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

    if config:
        data["config"] = config
    if description:
        data["description"] = description

    return await self.client.post("/1.0/storage-pools", data=data)

create_volume(pool_name, volume_name, volume_type, config=None) async

Create a new storage volume.

Parameters:

Name Type Description Default
pool_name str

Name of the storage pool.

required
volume_name str

Name of the volume.

required
volume_type str

Type of volume.

required
config Dict[str, Any]

Volume configuration.

None

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.

Source code in incus_sdk/api/storage_pools.py
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
async def create_volume(
    self,
    pool_name: str,
    volume_name: str,
    volume_type: str,
    config: Dict[str, Any] = None,
) -> Dict[str, Any]:
    """
    Create a new storage volume.

    Args:
        pool_name: Name of the storage pool.
        volume_name: Name of the volume.
        volume_type: Type of volume.
        config: Volume configuration.

    Returns:
        Dict[str, Any]: The operation response.
    """
    data = {"name": volume_name, "type": volume_type}

    if config:
        data["config"] = config

    return await self.client.post(
        f"/1.0/storage-pools/{pool_name}/volumes", data=data
    )

delete(name) async

Delete a storage pool.

Parameters:

Name Type Description Default
name str

Name of the storage pool.

required

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.

Source code in incus_sdk/api/storage_pools.py
109
110
111
112
113
114
115
116
117
118
119
async def delete(self, name: str) -> Dict[str, Any]:
    """
    Delete a storage pool.

    Args:
        name: Name of the storage pool.

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

delete_volume(pool_name, volume_name, volume_type) async

Delete a storage volume.

Parameters:

Name Type Description Default
pool_name str

Name of the storage pool.

required
volume_name str

Name of the volume.

required
volume_type str

Type of volume.

required

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.

Source code in incus_sdk/api/storage_pools.py
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
async def delete_volume(
    self, pool_name: str, volume_name: str, volume_type: str
) -> Dict[str, Any]:
    """
    Delete a storage volume.

    Args:
        pool_name: Name of the storage pool.
        volume_name: Name of the volume.
        volume_type: Type of volume.

    Returns:
        Dict[str, Any]: The operation response.
    """
    return await self.client.delete(
        f"/1.0/storage-pools/{pool_name}/volumes/{volume_type}/{volume_name}"
    )

get(name) async

Get a storage pool by name.

Parameters:

Name Type Description Default
name str

Name of the storage pool.

required

Returns:

Name Type Description
StoragePool StoragePool

The storage pool.

Source code in incus_sdk/api/storage_pools.py
42
43
44
45
46
47
48
49
50
51
52
53
async def get(self, name: str) -> StoragePool:
    """
    Get a storage pool by name.

    Args:
        name: Name of the storage pool.

    Returns:
        StoragePool: The storage pool.
    """
    response = await self.client.get(f"/1.0/storage-pools/{name}")
    return StoragePool(client=self, **response.get("metadata", {}))

get_volume(pool_name, volume_name, volume_type) async

Get a storage volume by name.

Parameters:

Name Type Description Default
pool_name str

Name of the storage pool.

required
volume_name str

Name of the volume.

required
volume_type str

Type of volume.

required

Returns:

Name Type Description
StorageVolume StorageVolume

The storage volume.

Source code in incus_sdk/api/storage_pools.py
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
async def get_volume(
    self, pool_name: str, volume_name: str, volume_type: str
) -> StorageVolume:
    """
    Get a storage volume by name.

    Args:
        pool_name: Name of the storage pool.
        volume_name: Name of the volume.
        volume_type: Type of volume.

    Returns:
        StorageVolume: The storage volume.
    """
    response = await self.client.get(
        f"/1.0/storage-pools/{pool_name}/volumes/{volume_type}/{volume_name}"
    )
    volume_data = response.get("metadata", {})
    volume_data["pool"] = pool_name
    return StorageVolume(client=self, **volume_data)

list(recursion=1) async

List all storage pools.

Parameters:

Name Type Description Default
recursion int

Level of recursion for the response.

1

Returns:

Type Description
List[StoragePool]

List[StoragePool]: List of storage pools.

Source code in incus_sdk/api/storage_pools.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[StoragePool]:
    """
    List all storage pools.

    Args:
        recursion: Level of recursion for the response.

    Returns:
        List[StoragePool]: List of storage pools.
    """
    params = {"recursion": recursion}
    response = await self.client.get("/1.0/storage-pools", params=params)

    pools = []
    for pool_data in response.get("metadata", []):
        pools.append(StoragePool(client=self, **pool_data))

    return pools

list_volumes(pool_name, volume_type=None, recursion=1) async

List volumes in a storage pool.

Parameters:

Name Type Description Default
pool_name str

Name of the storage pool.

required
volume_type str

Type of volumes to list.

None
recursion int

Level of recursion for the response.

1

Returns:

Type Description
List[StorageVolume]

List[StorageVolume]: List of storage volumes.

Source code in incus_sdk/api/storage_pools.py
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
async def list_volumes(
    self, pool_name: str, volume_type: str = None, recursion: int = 1
) -> List[StorageVolume]:
    """
    List volumes in a storage pool.

    Args:
        pool_name: Name of the storage pool.
        volume_type: Type of volumes to list.
        recursion: Level of recursion for the response.

    Returns:
        List[StorageVolume]: List of storage volumes.
    """
    path = f"/1.0/storage-pools/{pool_name}/volumes"
    if volume_type:
        path = f"{path}/{volume_type}"

    params = {"recursion": recursion}
    response = await self.client.get(path, params=params)

    volumes = []
    for volume_data in response.get("metadata", []):
        volume_data["pool"] = pool_name
        volumes.append(StorageVolume(client=self, **volume_data))

    return volumes

rename_volume(pool_name, volume_name, volume_type, new_name) async

Rename a storage volume.

Parameters:

Name Type Description Default
pool_name str

Name of the storage pool.

required
volume_name str

Current name of the volume.

required
volume_type str

Type of volume.

required
new_name str

New name for the volume.

required

Returns:

Type Description
Dict[str, Any]

Dict[str, Any]: The operation response.

Source code in incus_sdk/api/storage_pools.py
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
async def rename_volume(
    self, pool_name: str, volume_name: str, volume_type: str, new_name: str
) -> Dict[str, Any]:
    """
    Rename a storage volume.

    Args:
        pool_name: Name of the storage pool.
        volume_name: Current name of the volume.
        volume_type: Type of volume.
        new_name: New name for the volume.

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

    return await self.client.post(
        f"/1.0/storage-pools/{pool_name}/volumes/{volume_type}/{volume_name}",
        data=data,
    )

replace(name, config) async

Replace a storage pool configuration.

Parameters:

Name Type Description Default
name str

Name of the storage pool.

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/storage_pools.py
 96
 97
 98
 99
100
101
102
103
104
105
106
107
async def replace(self, name: str, config: Dict[str, Any]) -> Dict[str, Any]:
    """
    Replace a storage pool configuration.

    Args:
        name: Name of the storage pool.
        config: New configuration.

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

replace_volume(pool_name, volume_name, volume_type, config) async

Replace a storage volume configuration.

Parameters:

Name Type Description Default
pool_name str

Name of the storage pool.

required
volume_name str

Name of the volume.

required
volume_type str

Type of volume.

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/storage_pools.py
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
async def replace_volume(
    self, pool_name: str, volume_name: str, volume_type: str, config: Dict[str, Any]
) -> Dict[str, Any]:
    """
    Replace a storage volume configuration.

    Args:
        pool_name: Name of the storage pool.
        volume_name: Name of the volume.
        volume_type: Type of volume.
        config: New configuration.

    Returns:
        Dict[str, Any]: The operation response.
    """
    return await self.client.put(
        f"/1.0/storage-pools/{pool_name}/volumes/{volume_type}/{volume_name}",
        data=config,
    )

update(name, config) async

Update a storage pool.

Parameters:

Name Type Description Default
name str

Name of the storage pool.

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/storage_pools.py
83
84
85
86
87
88
89
90
91
92
93
94
async def update(self, name: str, config: Dict[str, Any]) -> Dict[str, Any]:
    """
    Update a storage pool.

    Args:
        name: Name of the storage pool.
        config: New configuration.

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

update_volume(pool_name, volume_name, volume_type, config) async

Update a storage volume.

Parameters:

Name Type Description Default
pool_name str

Name of the storage pool.

required
volume_name str

Name of the volume.

required
volume_type str

Type of volume.

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/storage_pools.py
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
async def update_volume(
    self, pool_name: str, volume_name: str, volume_type: str, config: Dict[str, Any]
) -> Dict[str, Any]:
    """
    Update a storage volume.

    Args:
        pool_name: Name of the storage pool.
        volume_name: Name of the volume.
        volume_type: Type of volume.
        config: New configuration.

    Returns:
        Dict[str, Any]: The operation response.
    """
    return await self.client.patch(
        f"/1.0/storage-pools/{pool_name}/volumes/{volume_type}/{volume_name}",
        data=config,
    )

Storage Pool Methods

list

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

List all storage pools.

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

Returns: - A list of StoragePool objects.

get

async def get(name: str) -> StoragePool

Get a storage pool by name.

Parameters: - name: Name of the storage pool.

Returns: - A StoragePool object.

create

async def create(
    name: str,
    driver: str,
    config: Dict[str, Any] = None,
    description: str = None,
) -> Dict[str, Any]

Create a new storage pool.

Parameters: - name: Name of the storage pool. - driver: Storage driver (e.g., "dir", "zfs", "btrfs", "lvm"). - config: Storage pool configuration as a dictionary (optional). - description: Description of the storage pool (optional).

Returns: - The operation response as a dictionary.

Example:

# Create a directory-backed storage pool
await client.storage_pools.create(
    name="my-dir-pool",
    driver="dir",
    config={
        "source": "/var/lib/incus/storage-pools/my-dir-pool"
    },
    description="Directory-backed storage pool"
)

# Create a ZFS storage pool
await client.storage_pools.create(
    name="my-zfs-pool",
    driver="zfs",
    config={
        "source": "my-zfs-pool/incus"
    },
    description="ZFS storage pool"
)

update

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

Update a storage pool.

Parameters: - name: Name of the storage pool. - config: New configuration as a dictionary.

Returns: - The operation response as a dictionary.

Note: This method performs a partial update of the storage pool configuration, only modifying the specified fields.

replace

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

Replace a storage pool configuration.

Parameters: - name: Name of the storage pool. - config: New configuration as a dictionary.

Returns: - The operation response as a dictionary.

Note: This method replaces the entire storage pool configuration with the provided configuration.

delete

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

Delete a storage pool.

Parameters: - name: Name of the storage pool.

Returns: - The operation response as a dictionary.

Storage Volume Methods

list_volumes

async def list_volumes(
    pool_name: str, 
    volume_type: str = None, 
    recursion: int = 1
) -> List[StorageVolume]

List volumes in a storage pool.

Parameters: - pool_name: Name of the storage pool. - volume_type: Type of volumes to list (optional). If not specified, all volume types are listed. - recursion: Level of recursion for the response (default: 1).

Returns: - A list of StorageVolume objects.

Example:

# List all volumes
all_volumes = await client.storage_pools.list_volumes("default")

# List only container volumes
container_volumes = await client.storage_pools.list_volumes("default", "container")

# List only custom volumes
custom_volumes = await client.storage_pools.list_volumes("default", "custom")

get_volume

async def get_volume(
    pool_name: str, 
    volume_name: str, 
    volume_type: str
) -> StorageVolume

Get a storage volume by name.

Parameters: - pool_name: Name of the storage pool. - volume_name: Name of the volume. - volume_type: Type of volume (e.g., "container", "virtual-machine", "image", "custom").

Returns: - A StorageVolume object.

create_volume

async def create_volume(
    pool_name: str,
    volume_name: str,
    volume_type: str,
    config: Dict[str, Any] = None,
) -> Dict[str, Any]

Create a new storage volume.

Parameters: - pool_name: Name of the storage pool. - volume_name: Name of the volume. - volume_type: Type of volume (e.g., "container", "virtual-machine", "image", "custom"). - config: Volume configuration as a dictionary (optional).

Returns: - The operation response as a dictionary.

Example:

# Create a custom volume
await client.storage_pools.create_volume(
    pool_name="default",
    volume_name="my-volume",
    volume_type="custom",
    config={
        "size": "10GB",
        "block.filesystem": "ext4"
    }
)

update_volume

async def update_volume(
    pool_name: str, 
    volume_name: str, 
    volume_type: str, 
    config: Dict[str, Any]
) -> Dict[str, Any]

Update a storage volume.

Parameters: - pool_name: Name of the storage pool. - volume_name: Name of the volume. - volume_type: Type of volume. - config: New configuration as a dictionary.

Returns: - The operation response as a dictionary.

Note: This method performs a partial update of the volume configuration, only modifying the specified fields.

replace_volume

async def replace_volume(
    pool_name: str, 
    volume_name: str, 
    volume_type: str, 
    config: Dict[str, Any]
) -> Dict[str, Any]

Replace a storage volume configuration.

Parameters: - pool_name: Name of the storage pool. - volume_name: Name of the volume. - volume_type: Type of volume. - config: New configuration as a dictionary.

Returns: - The operation response as a dictionary.

Note: This method replaces the entire volume configuration with the provided configuration.

delete_volume

async def delete_volume(
    pool_name: str, 
    volume_name: str, 
    volume_type: str
) -> Dict[str, Any]

Delete a storage volume.

Parameters: - pool_name: Name of the storage pool. - volume_name: Name of the volume. - volume_type: Type of volume.

Returns: - The operation response as a dictionary.

rename_volume

async def rename_volume(
    pool_name: str, 
    volume_name: str, 
    volume_type: str, 
    new_name: str
) -> Dict[str, Any]

Rename a storage volume.

Parameters: - pool_name: Name of the storage pool. - volume_name: Current name of the volume. - volume_type: Type of volume. - new_name: New name for the volume.

Returns: - The operation response as a dictionary.