Utilities and Helpers

from maas.client.utils.maas_async import asynchronous, Asynchronous, is_loop_running

def asynchronous(func):
    """
    Decorator that makes async functions callable synchronously when not in async context.

    Args:
        func: Async function to wrap

    Returns:
        Wrapped function that adapts to execution context
    """

class Asynchronous:
    """
    Metaclass for creating classes with automatic async/sync adaptation.

    Classes using this metaclass can have methods that work both synchronously
    and asynchronously depending on calling context.
    """

def is_loop_running():
    """
    Check if asyncio event loop is currently running.

    Returns:
        bool: True if event loop is running, False otherwise
    """

from maas.client.utils.maas_async import asynchronous, is_loop_running

@asynchronous
async def my_async_function():
    # Async implementation
    await some_async_operation()
    return result

# Call synchronously when not in async context
result = my_async_function()

# Call asynchronously when in async context
async def async_caller():
    result = await my_async_function()

from maas.client.utils import retries, gen_retries

def retries(timeout=30, intervals=1, time=time):
    """
    Decorator for retrying functions on failure.

    Args:
        timeout (float): Total timeout in seconds
        intervals (float): Interval between retries in seconds
        time: Time module for testing

    Returns:
        Decorator function
    """

def gen_retries(start, end, intervals, time=time):
    """
    Generate retry intervals.

    Args:
        start (float): Start time
        end (float): End time
        intervals (float): Interval between retries
        time: Time module

    Yields:
        None at each retry interval until timeout
    """

from maas.client.utils import retries

@retries(timeout=60, intervals=2)
def flaky_operation():
    # Will retry for up to 60 seconds with 2-second intervals
    if random.random() < 0.5:
        raise Exception("Temporary failure")
    return "success"

from maas.client.utils import api_url, ensure_trailing_slash, urlencode

def api_url(string):
    """
    Validate and normalize MAAS API URL.

    Args:
        string (str): URL to validate

    Returns:
        str: Normalized URL ending with /MAAS/

    Raises:
        ValueError: If URL is invalid
    """

def ensure_trailing_slash(string):
    """
    Ensure string ends with trailing slash.

    Args:
        string (str): String to process

    Returns:
        str: String with trailing slash
    """

def urlencode(data):
    """
    URL-encode data for HTTP requests.

    Args:
        data (dict): Data to encode

    Returns:
        str: URL-encoded string
    """

from maas.client.utils import prepare_payload

def prepare_payload(op, method, uri, data):
    """
    Prepare HTTP request payload.

    Args:
        op (str): Operation name
        method (str): HTTP method
        uri (str): Request URI
        data (dict): Request data

    Returns:
        tuple: (method, uri, data) prepared for request
    """

from maas.client.utils import parse_docstring

def parse_docstring(thing):
    """
    Parse Python docstrings.

    Args:
        thing: Object with __doc__ attribute

    Returns:
        dict: Parsed docstring information
    """

from maas.client.utils import get_all_subclasses, vars_class

def get_all_subclasses(cls):
    """
    Get all subclasses of a class recursively.

    Args:
        cls: Base class

    Returns:
        set: Set of all subclasses
    """

def vars_class(cls):
    """
    Get class variables (not instance variables).

    Args:
        cls: Class to inspect

    Returns:
        dict: Class variables
    """

from maas.client.utils import coalesce, remove_None

def coalesce(*values, default=None):
    """
    Return first non-None value, or default if all are None.

    Args:
        *values: Values to check
        default: Default value if all are None

    Returns:
        First non-None value or default
    """

def remove_None(params):
    """
    Remove None values from dictionary.

    Args:
        params (dict): Dictionary to filter

    Returns:
        dict: Dictionary with None values removed
    """

from maas.client.utils import coalesce, remove_None

# Coalesce
result = coalesce(None, None, "value", "other")  # Returns "value"

# Remove None values
params = {"a": 1, "b": None, "c": 3}
filtered = remove_None(params)  # {"a": 1, "c": 3}

from maas.client.utils.diff import calculate_dict_diff

def calculate_dict_diff(old_params, new_params):
    """
    Calculate differences between two dictionaries.

    Args:
        old_params (dict): Original dictionary
        new_params (dict): Modified dictionary

    Returns:
        dict: Dictionary containing only changed values
    """

from maas.client.utils import Spinner, SpinnerContext

class Spinner:
    """Terminal spinner for progress indication."""
    def start(self): ...
    def stop(self): ...

class SpinnerContext:
    """Context manager for spinner."""

from maas.client.utils import SpinnerContext

with SpinnerContext() as spinner:
    # Long-running operation
    perform_operation()
# Spinner stops automatically

from maas.client.utils.multipart import (
    get_content_type,
    make_bytes_payload,
    make_string_payload,
    make_file_payload,
    make_payloads,
    build_multipart_message,
    encode_multipart_message,
    encode_multipart_data
)

def encode_multipart_data(data=(), files=()):
    """
    Encode form data and files as multipart message.

    Args:
        data (tuple): Form data as (name, value) tuples
        files (tuple): Files as (name, file_object) tuples

    Returns:
        tuple: (content_type, encoded_data)
    """

from maas.client.utils.types import JSONValue, JSONArray, JSONObject

# Type aliases for type hints
JSONValue = ...   # str | int | float | bool | None | JSONArray | JSONObject
JSONArray = ...   # list[JSONValue]
JSONObject = ...  # dict[str, JSONValue]

from maas.client.errors import (
    MAASException,
    OperationNotAllowed,
    ObjectNotLoaded,
    CannotDelete,
    PowerError
)

class MAASException(Exception):
    """
    Base exception for all MAAS errors.

    Attributes:
        obj: The object that caused the exception
    """

class OperationNotAllowed(Exception):
    """Raised when MAAS says operation cannot be performed."""

class ObjectNotLoaded(Exception):
    """Raised when object data is not fully loaded."""

class CannotDelete(Exception):
    """Raised when object cannot be deleted."""

class PowerError(MAASException):
    """Raised when machine fails to power on or off."""

from maas.client.bones.helpers import (
    RemoteError,
    ConnectError,
    LoginError,
    PasswordWithoutUsername,
    UsernameWithoutPassword,
    LoginNotSupported,
    MacaroonLoginNotSupported
)

class RemoteError(Exception):
    """Raised when remote operation encounters an error."""

class ConnectError(Exception):
    """Raised when connection to MAAS server fails."""

class LoginError(Exception):
    """Raised when login operation fails."""

class PasswordWithoutUsername(LoginError):
    """Raised when password is provided without username."""

class UsernameWithoutPassword(LoginError):
    """Raised when username is provided without password."""

class LoginNotSupported(LoginError):
    """Raised when server does not support login-type authentication for API clients."""

class MacaroonLoginNotSupported(LoginError):
    """Raised when server does not support macaroon authentication for API clients."""

from maas.client.viscera.machines import (
    MachineNotFound,
    RescueModeFailure,
    FailedCommissioning,
    FailedTesting,
    FailedDeployment,
    FailedReleasing,
    FailedDiskErasing
)

class MachineNotFound(Exception):
    """Raised when no machine matching criteria is found."""

class RescueModeFailure(MAASException):
    """Raised when machine fails to perform rescue mode transition."""

class FailedCommissioning(MAASException):
    """Raised when machine fails to commission."""

class FailedTesting(MAASException):
    """Raised when machine fails testing."""

class FailedDeployment(MAASException):
    """Raised when machine fails to deploy."""

class FailedReleasing(MAASException):
    """Raised when machine fails to release."""

class FailedDiskErasing(MAASException):
    """Raised when machine fails to erase disks when releasing."""

from maas.client.flesh import CommandError

class CommandError(Exception):
    """Raised when CLI command execution fails."""

from maas.client.viscera.spaces import DeleteDefaultSpace

class DeleteDefaultSpace(Exception):
    """Raised when attempting to delete the default space."""

from maas.client import connect, login
from maas.client.errors import PowerError, OperationNotAllowed
from maas.client.bones.helpers import (
    ConnectError,
    LoginError,
    LoginNotSupported
)
from maas.client.viscera.machines import (
    MachineNotFound,
    FailedDeployment,
    FailedCommissioning
)
from maas.client.viscera.spaces import DeleteDefaultSpace

# Handle connection errors
try:
    client = connect('http://maas.example.com:5240/MAAS/', apikey='key')
except ConnectError as e:
    print(f"Connection failed: {e}")

# Handle login errors
try:
    client = login('http://maas.example.com:5240/MAAS/', username='admin', password='pass')
except LoginNotSupported as e:
    print(f"Login not supported: {e}")
except LoginError as e:
    print(f"Login failed: {e}")

# Handle allocation errors
try:
    machine = client.machines.allocate(cpu_count=8, memory=16384)
except MachineNotFound:
    print("No machine matching requirements found")

# Handle commissioning errors
try:
    machine.commission(wait=True)
except FailedCommissioning as e:
    print(f"Commissioning failed: {e}")
    print(f"Machine: {e.obj.hostname}")

# Handle deployment errors
try:
    machine.deploy(distro_series='jammy', wait=True)
except FailedDeployment as e:
    print(f"Deployment failed: {e}")
    print(f"Machine status: {e.obj.status}")

# Handle power errors
try:
    machine.power_on()
except PowerError as e:
    print(f"Power operation failed: {e}")
except OperationNotAllowed as e:
    print(f"Operation not allowed: {e}")

# Handle space deletion errors
try:
    default_space = client.spaces.get(0)
    await default_space.delete()
except DeleteDefaultSpace as e:
    print(f"Cannot delete default space: {e}")