tessl install tessl/pypi-python-libmaas@0.6.0Python client library for MAAS 2.0+ with sync/async support, providing machine provisioning, network management, and storage configuration.
Manage physical servers through MAAS, including allocation, deployment, commissioning, power control, storage, and network configuration.
Retrieve and filter machines from MAAS.
# Via Client facade
client.machines.list()from maas.client import connect
client = connect('http://maas.example.com:5240/MAAS/', apikey='key')
# List all machines
machines = client.machines.list()
for machine in machines:
print(f"{machine.hostname}: {machine.status}")
# Machines have properties like:
# - system_id: Unique identifier
# - hostname: Machine hostname
# - status: Current status (NodeStatus enum)
# - architecture: CPU architecture
# - memory: RAM in MB
# - cpu_count: Number of CPUs
# - power_state: Current power state
# - ip_addresses: List of IP addressesRetrieve a single machine by system ID or other criteria.
client.machines.get(system_id)
client.machines.get(**filters)# Get by system ID
machine = client.machines.get('abc123')
# Get by hostname
machine = client.machines.get(hostname='node1.maas')Allocate available machines based on criteria like architecture, memory, CPU count, tags, etc.
client.machines.allocate(**criteria)Parameters for allocation:
name (str): Hostname for the allocated machinearchitecture (str): Architecture constraint (e.g., 'amd64/generic')cpu_count (int): Minimum CPU countmem (int): Minimum memory in MBtags (list): Required tagszone (str): Availability zonepool (str): Resource poolcomment (str): Allocation comment# Allocate any available machine
machine = client.machines.allocate()
# Allocate with constraints
machine = client.machines.allocate(
architecture='amd64/generic',
cpu_count=4,
mem=8192, # 8 GB
tags=['ssd', 'compute'],
zone='zone1',
comment='Web server allocation'
)
print(f"Allocated: {machine.hostname} ({machine.system_id})")Register new machines with MAAS.
client.machines.create(**params)Parameters:
hostname (str): Machine hostnamearchitecture (str): Machine architecturemac_addresses (list): List of MAC addressespower_type (str): Power management type (e.g., 'ipmi', 'virsh', 'manual')power_parameters (dict): Power-specific parametersdomain (str): DNS domainpool (str): Resource pool# Create machine with IPMI power control
machine = client.machines.create(
hostname='newnode',
architecture='amd64/generic',
mac_addresses=['00:11:22:33:44:55'],
power_type='ipmi',
power_parameters={
'power_address': '192.168.1.100',
'power_user': 'admin',
'power_pass': 'password'
}
)Deploy operating systems to allocated or ready machines.
machine.deploy(
distro_series=None,
hwe_kernel=None,
user_data=None,
comment=None,
wait=False,
install_kvm=False,
wait_interval=5
)Parameters:
distro_series (str, optional): OS release to deploy (e.g., 'focal', 'jammy')hwe_kernel (str, optional): Hardware enablement kerneluser_data (bytes or str, optional): Cloud-init user data (base64 encoded if str, auto-encoded if bytes)comment (str, optional): Deployment commentwait (bool, optional): Wait until deployment completesinstall_kvm (bool, optional): Install KVM on the machinewait_interval (int, optional): Polling interval in seconds when waiting# Allocate and deploy
machine = client.machines.allocate(cpu_count=2, mem=4096)
machine.deploy(distro_series='jammy')
# Deploy with custom kernel
machine.deploy(
distro_series='focal',
hwe_kernel='hwe-20.04',
comment='Production web server'
)
# Deploy with cloud-init user data
user_data = """#cloud-config
packages:
- nginx
- postgresql
"""
machine.deploy(distro_series='jammy', user_data=user_data)Commission machines to gather hardware details and run tests.
machine.commission(
enable_ssh=None,
skip_networking=None,
skip_storage=None,
commissioning_scripts=None,
testing_scripts=None,
wait=False,
wait_interval=5
)Parameters:
enable_ssh (bool, optional): Enable SSH during commissioningskip_networking (bool, optional): Skip network configurationskip_storage (bool, optional): Skip storage configurationcommissioning_scripts (list, optional): List of commissioning scripts to runtesting_scripts (list, optional): List of test scripts to run (pass empty list or "none" to disable)wait (bool, optional): Wait until commissioning completeswait_interval (int, optional): Polling interval in seconds when waiting# Basic commissioning
machine.commission()
# Commission with SSH enabled
machine.commission(enable_ssh=True)
# Commission with specific tests
machine.commission(
testing_scripts=['smartctl-validate', 'memtester'],
skip_networking=False
)Release allocated machines back to the pool.
machine.release(
comment=None,
erase=None,
secure_erase=None,
quick_erase=None,
wait=False,
wait_interval=5
)Parameters:
comment (str, optional): Release commenterase (bool, optional): Erase the disk when releasing (deprecated, use secure_erase or quick_erase)secure_erase (bool, optional): Use drive's secure erase feature if availablequick_erase (bool, optional): Quick erase (wipe beginning and end only, not secure)wait (bool, optional): Wait until release completeswait_interval (int, optional): Polling interval in seconds when waiting# Release machine
machine.release(comment='No longer needed')
# Release with secure erase
machine.release(secure_erase=True)
# Release with quick erase (fast but not secure)
machine.release(quick_erase=True)Control machine power state.
machine.power_on(comment=None, wait=False, wait_interval=5)
machine.power_off(stop_mode=PowerStopMode.HARD, comment=None, wait=False, wait_interval=5)
machine.query_power_state()# Power on machine
machine.power_on()
# Power on with comment and wait
machine.power_on(comment='Starting maintenance', wait=True)
# Power off machine (soft shutdown)
from maas.client.enum import PowerStopMode
machine.power_off(stop_mode=PowerStopMode.SOFT)
# Power off machine (hard power off)
machine.power_off(stop_mode=PowerStopMode.HARD, comment='Emergency shutdown', wait=True)
# Query current power state
state = machine.query_power_state()
print(f"Power state: {state}") # PowerState.ON, PowerState.OFF, etc.Lock machines to prevent unintended changes.
machine.lock(comment=None)
machine.unlock(comment=None)# Lock machine
machine.lock(comment='Maintenance in progress')
# Unlock machine
machine.unlock()Abort ongoing machine operations.
machine.abort(comment=None)# Abort deployment or commissioning
machine.abort(comment='Configuration error, restarting')Enter or exit rescue mode for system recovery.
machine.enter_rescue_mode(wait=False, wait_interval=5)
machine.exit_rescue_mode(wait=False, wait_interval=5)# Enter rescue mode
machine.enter_rescue_mode()
# Enter rescue mode and wait for completion
machine.enter_rescue_mode(wait=True)
# Perform recovery operations...
# Exit rescue mode
machine.exit_rescue_mode()
# Exit rescue mode and wait for completion
machine.exit_rescue_mode(wait=True)Mark machines as broken or fixed.
machine.mark_broken(comment=None)
machine.mark_fixed(comment=None)# Mark as broken
machine.mark_broken(comment='Hardware failure detected')
# Mark as fixed after repair
machine.mark_fixed(comment='Hardware replaced')Get and restore machine configurations.
machine.get_curtin_config()
machine.get_details()
machine.restore_default_configuration()
machine.restore_networking_configuration()
machine.restore_storage_configuration()# Get deployment configuration (Curtin preseed data)
curtin_config = machine.get_curtin_config()
# Returns dict with cloud-init and curtin configuration used during deployment
# Get detailed machine information
details = machine.get_details()
# Returns dict with comprehensive machine details including:
# - lldp and lshw discovered hardware information
# - Storage layout and partition details
# - Network topology and link information
# - Power configuration details
# Restore all configuration to defaults
machine.restore_default_configuration()
# Resets networking and storage to initial state
# Restore only networking configuration
machine.restore_networking_configuration()
# Removes custom network configuration, resets to auto-discovery
# Restore only storage configuration
machine.restore_storage_configuration()
# Removes custom storage layout, resets to default partitioningClear default gateways on machine interfaces.
machine.clear_default_gateways()# Clear all default gateways
machine.clear_default_gateways()Modify and save machine attributes.
machine.save()
machine.refresh()# Update machine properties
machine.hostname = 'new-hostname'
machine.description = 'Updated description'
machine.save()
# Refresh from server
machine.refresh()Handle machine operation errors with specific exceptions.
from maas.client.errors import (
PowerError,
MAASException,
OperationNotAllowed
)Common exceptions when working with machines:
PowerError: Raised when power operations fail (power on/off)OperationNotAllowed: Raised when an operation cannot be performed in the current machine stateMAASException: Base exception for general MAAS errorsfrom maas.client import connect
from maas.client.errors import PowerError, OperationNotAllowed, MAASException
client = connect('http://maas.example.com:5240/MAAS/', apikey='key')
try:
machine = client.machines.allocate(cpu_count=32, mem=65536)
machine.power_on(wait=True, wait_interval=5)
machine.deploy(wait=True)
except PowerError as e:
print(f"Power management failed: {e}")
# Handle power errors - check BMC configuration
except OperationNotAllowed as e:
print(f"Operation not allowed: {e}")
# Operation not valid for machine's current state
print(f"Machine status: {machine.status}")
except MAASException as e:
print(f"MAAS error: {e}")
# General MAAS errors
# Check machine state before operations
from maas.client.enum import NodeStatus
if machine.status == NodeStatus.READY:
machine.deploy()
elif machine.status == NodeStatus.ALLOCATED:
print("Machine already allocated")
elif machine.status == NodeStatus.DEPLOYED:
print("Machine already deployed")Retrieve power configuration templates for machine types.
client.machines.get_power_parameters_for()# Get power parameters schema
params = client.machines.get_power_parameters_for()
# Returns available power types and their required parametersConvert generic Node objects to specific typed objects (Machine, Device, RackController, RegionController).
node.as_machine()
node.as_device()
node.as_rack_controller()
node.as_region_controller()When working with nodes from generic listings or when you need to access type-specific methods, use these conversion methods:
from maas.client.viscera import Nodes
# List all nodes (returns mixed types)
nodes = Nodes.read()
for node in nodes:
# Convert to specific type based on node_type
if node.node_type == NodeType.MACHINE:
machine = node.as_machine()
# Now can use machine-specific methods like deploy()
print(f"Machine: {machine.hostname}")
elif node.node_type == NodeType.DEVICE:
device = node.as_device()
print(f"Device: {device.hostname}")
elif node.node_type == NodeType.RACK_CONTROLLER:
controller = node.as_rack_controller()
print(f"Rack Controller: {controller.hostname}")Get and set power parameters for machines.
node.get_power_parameters()
node.set_power(power_type, power_parameters)Parameters for set_power:
power_type (str): Power management type (e.g., 'ipmi', 'virsh', 'manual', 'amt', 'wedge')power_parameters (dict): Type-specific parametersfrom maas.client import connect
client = connect('http://maas.example.com:5240/MAAS/', apikey='key')
machine = client.machines.get('abc123')
# Get current power parameters
params = machine.get_power_parameters()
print(f"Power type: {params['power_type']}")
print(f"Parameters: {params['power_parameters']}")
# Set IPMI power management
machine.set_power(
power_type='ipmi',
power_parameters={
'power_address': '192.168.1.100',
'power_user': 'admin',
'power_pass': 'password',
'power_driver': 'LAN_2_0',
'mac_address': '00:11:22:33:44:55'
}
)
# Set Virsh (libvirt) power management
machine.set_power(
power_type='virsh',
power_parameters={
'power_address': 'qemu+ssh://user@192.168.1.50/system',
'power_id': 'vm-name'
}
)
# Set manual power management (no automatic control)
machine.set_power(
power_type='manual',
power_parameters={}
)Machines have the following key properties:
system_id (str): Unique system identifierhostname (str): Machine hostnamefqdn (str): Fully qualified domain namestatus (NodeStatus, readonly): Current machine statusstatus_name (str, readonly): Human-readable status namestatus_action (str, readonly): Current status actionstatus_message (str, readonly): Status messagepower_state (PowerState): Current power statepower_type (str): Power management typearchitecture (str): CPU architecturecpus (int): Number of CPUsmemory (int): RAM in MBstorage (int): Total storage in bytesip_addresses (list): List of IP addressespool (ResourcePool): Resource poolzone (Zone): Availability zonedomain (Domain): DNS domainowner (User): Machine ownertags (list): List of tagsboot_disk (BlockDevice, readonly): Boot disk deviceboot_interface (Interface, readonly): Boot network interfaceblock_devices (BlockDevices): Collection of block devicesinterfaces (Interfaces): Network interfacesbcaches (Bcaches): Bcache devicescache_sets (BcacheCacheSets): Bcache cache setsraids (Raids): RAID arraysvolume_groups (VolumeGroups): LVM volume groupsdistro_series (str, readonly): Deployed OS distribution seriesosystem (str, readonly): Operating systemhwe_kernel (str): Hardware enablement kernelmin_hwe_kernel (str): Minimum HWE kernelnetboot (bool, readonly): Whether machine network bootsdisable_ipv4 (bool): Disable IPv4locked (bool, readonly): Whether machine is lockedowner_data (dict): Custom owner data key-value pairsfrom maas.client.enum import NodeStatus, PowerState, PowerStopMode
class NodeStatus:
"""Machine status enumeration."""
DEFAULT = ... # Alias for NEW
NEW = ...
COMMISSIONING = ...
FAILED_COMMISSIONING = ...
MISSING = ...
READY = ...
RESERVED = ...
ALLOCATED = ...
DEPLOYING = ...
DEPLOYED = ...
RETIRED = ...
BROKEN = ...
FAILED_DEPLOYMENT = ...
RELEASING = ...
FAILED_RELEASING = ...
DISK_ERASING = ...
FAILED_DISK_ERASING = ...
RESCUE_MODE = ...
ENTERING_RESCUE_MODE = ...
FAILED_ENTERING_RESCUE_MODE = ...
EXITING_RESCUE_MODE = ...
FAILED_EXITING_RESCUE_MODE = ...
TESTING = ...
FAILED_TESTING = ...
class PowerState:
"""Power state enumeration."""
ON = ...
OFF = ...
UNKNOWN = ...
ERROR = ...
class PowerStopMode:
"""Power stop mode enumeration."""
SOFT = ... # Graceful shutdown
HARD = ... # Immediate power off