Operations API

The OperationsAPI client provides methods for managing Incus operations.

Usage

from incus_sdk import Client

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

    # Get an operation by ID
    operation = await client.operations.get("operation-uuid")

    # Wait for an operation to complete
    result = await client.operations.wait("operation-uuid", timeout=30)

    # Cancel an operation
    await client.operations.cancel("operation-uuid")

    # Delete an operation
    await client.operations.delete("operation-uuid")

Class Documentation

API client for Incus operations.

Source code in incus_sdk/api/operations.py

class OperationsAPI:
    """API client for Incus operations."""

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

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

    async def list(self, recursion: int = 1) -> List[Dict[str, Any]]:
        """
        List all operations.

        Args:
            recursion: Level of recursion for the response.

        Returns:
            List[Dict[str, Any]]: List of operations.
        """
        params = {"recursion": recursion}
        response = await self.client.get("/1.0/operations", params=params)
        return response.get("metadata", [])

    async def get(self, operation_id: str) -> Dict[str, Any]:
        """
        Get an operation by ID.

        Args:
            operation_id: ID of the operation.

        Returns:
            Dict[str, Any]: The operation.
        """
        response = await self.client.get(f"/1.0/operations/{operation_id}")
        return response.get("metadata", {})

    async def delete(self, operation_id: str) -> Dict[str, Any]:
        """
        Delete an operation.

        Args:
            operation_id: ID of the operation.

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

    async def wait(self, operation_id: str, timeout: int = 60) -> Dict[str, Any]:
        """
        Wait for an operation to complete.

        Args:
            operation_id: ID of the operation.
            timeout: Timeout in seconds.

        Returns:
            Dict[str, Any]: The operation result.
        """
        params = {"timeout": timeout}
        response = await self.client.get(
            f"/1.0/operations/{operation_id}/wait", params=params
        )
        return response.get("metadata", {})

    async def cancel(self, operation_id: str) -> Dict[str, Any]:
        """
        Cancel an operation.

        Args:
            operation_id: ID of the operation.

        Returns:
            Dict[str, Any]: The operation response.
        """
        return await self.client.put(f"/1.0/operations/{operation_id}/cancel", data={})

__init__(client)

Initialize a new OperationsAPI client.

Parameters:

Name	Type	Description	Default
client	APIClient	The base API client.	required

Source code in incus_sdk/api/operations.py

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

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

cancel(operation_id) async

Cancel an operation.

Parameters:

Name	Type	Description	Default
operation_id	str	ID of the operation.	required

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: The operation response.

Source code in incus_sdk/api/operations.py

async def cancel(self, operation_id: str) -> Dict[str, Any]:
    """
    Cancel an operation.

    Args:
        operation_id: ID of the operation.

    Returns:
        Dict[str, Any]: The operation response.
    """
    return await self.client.put(f"/1.0/operations/{operation_id}/cancel", data={})

delete(operation_id) async

Delete an operation.

Parameters:

Name	Type	Description	Default
operation_id	str	ID of the operation.	required

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: The operation response.

Source code in incus_sdk/api/operations.py

async def delete(self, operation_id: str) -> Dict[str, Any]:
    """
    Delete an operation.

    Args:
        operation_id: ID of the operation.

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

get(operation_id) async

Get an operation by ID.

Parameters:

Name	Type	Description	Default
operation_id	str	ID of the operation.	required

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: The operation.

Source code in incus_sdk/api/operations.py

async def get(self, operation_id: str) -> Dict[str, Any]:
    """
    Get an operation by ID.

    Args:
        operation_id: ID of the operation.

    Returns:
        Dict[str, Any]: The operation.
    """
    response = await self.client.get(f"/1.0/operations/{operation_id}")
    return response.get("metadata", {})

list(recursion=1) async

List all operations.

Parameters:

Name	Type	Description	Default
recursion	int	Level of recursion for the response.	1

Returns:

Type	Description
List[Dict[str, Any]]	List[Dict[str, Any]]: List of operations.

Source code in incus_sdk/api/operations.py

async def list(self, recursion: int = 1) -> List[Dict[str, Any]]:
    """
    List all operations.

    Args:
        recursion: Level of recursion for the response.

    Returns:
        List[Dict[str, Any]]: List of operations.
    """
    params = {"recursion": recursion}
    response = await self.client.get("/1.0/operations", params=params)
    return response.get("metadata", [])

wait(operation_id, timeout=60) async

Wait for an operation to complete.

Parameters:

Name	Type	Description	Default
operation_id	str	ID of the operation.	required
timeout	int	Timeout in seconds.	60

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: The operation result.

Source code in incus_sdk/api/operations.py

async def wait(self, operation_id: str, timeout: int = 60) -> Dict[str, Any]:
    """
    Wait for an operation to complete.

    Args:
        operation_id: ID of the operation.
        timeout: Timeout in seconds.

    Returns:
        Dict[str, Any]: The operation result.
    """
    params = {"timeout": timeout}
    response = await self.client.get(
        f"/1.0/operations/{operation_id}/wait", params=params
    )
    return response.get("metadata", {})

Methods

list

async def list(recursion: int = 1) -> List[Dict[str, Any]]

List all operations.

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

Returns: - A list of operations as dictionaries.

get

async def get(operation_id: str) -> Dict[str, Any]

Get an operation by ID.

Parameters: - operation_id: ID of the operation.

Returns: - The operation as a dictionary.

Example:

# Get an operation and check its status
operation = await client.operations.get("operation-uuid")
print(f"Operation status: {operation.get('status')}")
print(f"Operation type: {operation.get('class')}")
print(f"Operation created at: {operation.get('created_at')}")

delete

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

Delete an operation.

Parameters: - operation_id: ID of the operation.

Returns: - The operation response as a dictionary.

wait

async def wait(
    operation_id: str, 
    timeout: int = 60
) -> Dict[str, Any]

Wait for an operation to complete.

Parameters: - operation_id: ID of the operation. - timeout: Timeout in seconds (default: 60).

Returns: - The operation result as a dictionary.

Example:

# Create an instance and wait for the operation to complete
response = await client.instances.create(
    name="my-instance",
    source={
        "type": "image",
        "alias": "ubuntu/22.04"
    }
)

# Wait for the operation to complete
operation_id = response.get("metadata", {}).get("id")
if operation_id:
    result = await client.operations.wait(operation_id, timeout=120)
    if result.get("status") == "Success":
        print("Instance created successfully")
    else:
        print(f"Failed to create instance: {result.get('err')}")

cancel

async def cancel(operation_id: str) -> Dict[str, Any]

Cancel an operation.

Parameters: - operation_id: ID of the operation.

Returns: - The operation response as a dictionary.

Note: Not all operations can be canceled. Only long-running operations that support cancellation can be canceled.