tessl/pypi-apache-libcloud

A standard Python library that abstracts away differences among multiple cloud provider APIs

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Compute Services

Name: tessl/pypi-apache-libcloud
Author: tessl

The compute service provides a unified interface for managing virtual machines, instances, and compute resources across 60+ cloud providers including AWS EC2, Azure, Google Compute Engine, Rackspace, Linode, DigitalOcean, and many others.

Providers

from libcloud.compute.types import Provider

class Provider:
    """Enumeration of supported compute providers"""
    # Major cloud providers
    AZURE = 'azure'
    AZURE_ARM = 'azure_arm'
    EC2 = 'ec2'
    GCE = 'gce'
    
    # Other providers (alphabetically)
    ABIQUO = 'abiquo'
    ALIYUN_ECS = 'aliyun_ecs'
    AURORACOMPUTE = 'aurora_compute'
    BRIGHTBOX = 'brightbox'
    CLOUDSCALE = 'cloudscale'
    CLOUDSTACK = 'cloudstack'
    DIGITAL_OCEAN = 'digitalocean'
    DIMENSIONDATA = 'dimensiondata'
    EQUINIXMETAL = 'equinixmetal'
    EXOSCALE = 'exoscale'
    GRIDSCALE = 'gridscale'
    IBM = 'ibm'
    IKOULA = 'ikoula'
    KAMATERA = 'kamatera'
    KTUCLOUD = 'ktucloud'
    KUBEVIRT = 'kubevirt'
    LIBVIRT = 'libvirt'
    LINODE = 'linode'
    MAXIHOST = 'maxihost'
    NIMBUS = 'nimbus'
    NINEFOLD = 'ninefold'
    NTTCIS = 'nttcis'
    OPENNEBULA = 'opennebula'
    OPENSTACK = 'openstack'
    OUTSCALE = 'outscale'
    OVH = 'ovh'
    RACKSPACE = 'rackspace'
    SCALEWAY = 'scaleway'
    UPCLOUD = 'upcloud'
    VCLOUD = 'vcloud'
    VSPHERE = 'vsphere'
    VULTR = 'vultr'
    # ... and 20+ more providers

Driver Factory

from libcloud.compute.providers import get_driver

def get_driver(provider: Provider) -> type[NodeDriver]

Get the driver class for a specific compute provider.

Parameters:

provider: Provider identifier from the Provider enum

Returns:

Driver class for the specified provider

Example:

from libcloud.compute.types import Provider
from libcloud.compute.providers import get_driver

# Get EC2 driver class
cls = get_driver(Provider.EC2)

# Initialize driver with credentials
driver = cls('access_key', 'secret_key', region='us-east-1')

Core Classes

NodeDriver

class NodeDriver(BaseDriver):
    """Base class for all compute drivers"""
    
    def list_nodes(self) -> List[Node]
    def get_node(self, node_id: str) -> Node
    def create_node(self, name: str, size: NodeSize, image: NodeImage, location: NodeLocation = None, **kwargs) -> Node
    def reboot_node(self, node: Node) -> bool
    def destroy_node(self, node: Node) -> bool
    def list_images(self, location: NodeLocation = None, ex_only_active: bool = True) -> List[NodeImage]
    def list_sizes(self, location: NodeLocation = None) -> List[NodeSize]
    def list_locations(self) -> List[NodeLocation]
    def deploy_node(self, deploy: Deployment, ssh_username: str = "root", 
                    ssh_port: int = 22, ssh_timeout: int = 10, timeout: int = 300, 
                    **create_node_kwargs) -> Node
    def wait_until_running(self, nodes: List[Node], wait_period: float = 5, timeout: int = 600, 
                           ssh_interface: str = "public_ips", force_ipv4: bool = True) -> List[Tuple[Node, List[str]]]

Base class that all compute drivers inherit from. Provides methods for managing virtual machines and instances.

Key Methods:

list_nodes(): List all nodes/instances in the account
create_node(): Create a new virtual machine instance
destroy_node(): Terminate/delete a node
reboot_node(): Restart a node
list_images(): List available OS images/templates
list_sizes(): List available instance sizes/flavors
list_locations(): List available regions/datacenters

Node

class Node:
    """Represents a virtual machine/compute instance"""
    
    id: str
    name: str
    state: NodeState
    public_ips: List[str]
    private_ips: List[str]
    driver: NodeDriver
    size: NodeSize
    image: NodeImage
    extra: Dict[str, Any]
    created_at: datetime
    
    def reboot(self) -> bool
    def destroy(self) -> bool
    def get_uuid(self) -> str

Represents a virtual machine or compute instance.

Properties:

id: Unique identifier for the node
name: Human-readable name
state: Current state (running, stopped, pending, etc.)
public_ips: List of public IP addresses
private_ips: List of private IP addresses
size: Instance size/flavor information
image: OS image information
extra: Provider-specific metadata

NodeSize

class NodeSize:
    """Represents an instance size/flavor"""
    
    id: str
    name: str
    ram: int
    disk: int
    bandwidth: int
    price: float
    driver: NodeDriver
    extra: Dict[str, Any]

Represents available instance sizes/flavors with their specifications.

Properties:

ram: Memory in MB
disk: Disk space in GB
bandwidth: Network bandwidth
price: Hourly cost

NodeImage

class NodeImage:
    """Represents an OS image/template"""
    
    id: str
    name: str
    driver: NodeDriver
    extra: Dict[str, Any]

Represents an operating system image or template that can be used to create instances.

NodeLocation

class NodeLocation:
    """Represents a geographic location/region"""
    
    id: str
    name: str
    country: str
    availability_zone: str
    driver: NodeDriver
    extra: Dict[str, Any]

Represents a geographic location, region, or availability zone where instances can be created.

NodeState

class NodeState:
    """Node lifecycle states"""
    RUNNING = 0
    REBOOTING = 1
    TERMINATED = 2
    PENDING = 3
    UNKNOWN = 4
    STOPPED = 5
    SUSPENDED = 6
    ERROR = 7
    PAUSED = 8
    RECONFIGURING = 9
    MIGRATING = 10
    STARTING = 11
    STOPPING = 12

Enumeration of possible node states during their lifecycle.

Authentication

NodeAuth Classes

class NodeAuthSSHKey:
    """SSH key authentication"""
    def __init__(self, pubkey: str)
    
class NodeAuthPassword:
    """Password authentication"""  
    def __init__(self, password: str)

Authentication methods for accessing created nodes.

Example:

from libcloud.compute.base import NodeAuthSSHKey, NodeAuthPassword

# SSH key authentication
ssh_key = NodeAuthSSHKey(open('~/.ssh/id_rsa.pub').read())

# Password authentication  
password_auth = NodeAuthPassword('my_secure_password')

# Use in node creation
node = driver.create_node(
    name='my-server',
    size=sizes[0],
    image=images[0],
    auth=ssh_key
)

Storage Volumes

StorageVolume

class StorageVolume(UuidMixin):
    """Represents a storage volume"""
    
    def __init__(self, id: str, name: str, size: int, driver: NodeDriver, 
                 state: StorageVolumeState = None, extra: Dict = None) -> None
    
    # Instance methods
    def list_snapshots(self) -> List[VolumeSnapshot]
    def attach(self, node: Node, device: str = None) -> bool
    def detach(self) -> bool
    def snapshot(self, name: str) -> VolumeSnapshot  
    def destroy(self) -> bool
    
    # Attributes
    id: str
    name: str
    size: int
    state: StorageVolumeState
    driver: NodeDriver
    extra: Dict[str, Any]

VolumeSnapshot

class VolumeSnapshot:
    """Represents a volume snapshot"""
    
    def __init__(self, id: str, driver: NodeDriver, size: int = None, 
                 extra: Dict = None, created: datetime = None, 
                 state: StorageVolumeState = None, name: str = None) -> None
    
    def destroy(self) -> bool
    
    # Attributes
    id: str
    name: str
    size: int
    driver: NodeDriver
    state: StorageVolumeState
    created: datetime
    extra: Dict[str, Any]

Volume Management Methods

class NodeDriver:
    def list_volumes(self) -> List[StorageVolume]
    def create_volume(self, size: int, name: str, location: NodeLocation = None, snapshot: VolumeSnapshot = None) -> StorageVolume
    def destroy_volume(self, volume: StorageVolume) -> bool
    def attach_volume(self, node: Node, volume: StorageVolume, device: str = None) -> bool
    def detach_volume(self, volume: StorageVolume) -> bool
    def list_volume_snapshots(self, volume: StorageVolume) -> List[VolumeSnapshot]
    def create_volume_snapshot(self, volume: StorageVolume, name: str = None) -> VolumeSnapshot
    def destroy_volume_snapshot(self, snapshot: VolumeSnapshot) -> bool

SSH Key Management

KeyPair

class KeyPair:
    """Represents an SSH key pair"""
    
    name: str
    fingerprint: str
    public_key: str
    private_key: str
    driver: NodeDriver
    extra: Dict[str, Any]

Key Management Methods

class NodeDriver:
    def list_key_pairs(self) -> List[KeyPair]
    def create_key_pair(self, name: str) -> KeyPair
    def import_key_pair_from_string(self, name: str, key_material: str) -> KeyPair
    def import_key_pair_from_file(self, name: str, key_file_path: str) -> KeyPair
    def delete_key_pair(self, key_pair: KeyPair) -> bool

SSH Client Support

from libcloud.compute.ssh import SSHClient

class SSHClient:
    """SSH client for connecting to nodes"""
    
    def __init__(self, hostname: str, port: int = 22, username: str = 'root', password: str = None, key: str = None)
    def connect(self) -> bool
    def run(self, cmd: str) -> Tuple[str, str, int]
    def put(self, path: str, contents: str, chmod: int = 755) -> bool
    def delete(self, path: str) -> bool
    def close(self) -> bool

Provides SSH connectivity to created nodes for remote command execution.

Example:

from libcloud.compute.ssh import SSHClient

# Connect via SSH key
client = SSHClient(
    hostname=node.public_ips[0],
    username='ubuntu',
    key='/path/to/private_key'
)

if client.connect():
    stdout, stderr, exit_code = client.run('ls -la')
    print(f"Command output: {stdout}")
    client.close()

Deployment Classes

Deployment classes provide automated server provisioning and configuration capabilities. They work with the deploy_node() method to set up servers after creation.

Base Deployment Class

from libcloud.compute.deployment import Deployment

class Deployment:
    """Base class for all deployment tasks"""
    
    def run(self, node: Node, client: BaseSSHClient) -> Node

Abstract base class that all deployment steps inherit from.

SSH Key Deployment

from libcloud.compute.deployment import SSHKeyDeployment

class SSHKeyDeployment(Deployment):
    """Installs SSH public key for authentication"""
    
    def __init__(self, key: Union[str, IO]) -> None
    def run(self, node: Node, client: BaseSSHClient) -> Node

Installs a public SSH key into the server's authorized_keys file.

Parameters:

key: SSH public key content as string or file object

Example:

from libcloud.compute.deployment import SSHKeyDeployment

# From string
key_deploy = SSHKeyDeployment("ssh-rsa AAAAB3NzaC1yc2E...")

# From file
with open('/path/to/key.pub') as fp:
    key_deploy = SSHKeyDeployment(fp)

File Deployment

from libcloud.compute.deployment import FileDeployment

class FileDeployment(Deployment):
    """Transfers files to the server"""
    
    def __init__(self, source: str, target: str) -> None
    def run(self, node: Node, client: BaseSSHClient) -> Node

Uploads a local file to the remote server.

Parameters:

source: Local file path
target: Remote destination path

Example:

from libcloud.compute.deployment import FileDeployment

# Transfer configuration file
config_deploy = FileDeployment("/local/app.conf", "/etc/app/app.conf")

Script Deployment

from libcloud.compute.deployment import ScriptDeployment

class ScriptDeployment(Deployment):
    """Executes shell scripts on the server"""
    
    def __init__(self, script: str, args: List[str] = None, name: str = None, 
                 delete: bool = False, timeout: float = None) -> None
    def run(self, node: Node, client: BaseSSHClient) -> Node
    
    # Available after execution
    stdout: str
    stderr: str
    exit_status: int

Uploads and executes a shell script on the server.

Parameters:

script: Script content as string
args: Optional command line arguments
name: Optional script filename (random if not specified)
delete: Whether to delete script after execution (default: False)
timeout: Optional execution timeout in seconds

Example:

from libcloud.compute.deployment import ScriptDeployment

# Basic script
install_script = ScriptDeployment('''#!/bin/bash
apt-get update
apt-get install -y nginx
systemctl start nginx
''')

# Script with arguments and timeout
config_script = ScriptDeployment(
    script="#!/bin/bash\necho $1 > /etc/hostname",
    args=["web-server"],
    delete=True,
    timeout=30
)

Script File Deployment

from libcloud.compute.deployment import ScriptFileDeployment

class ScriptFileDeployment(ScriptDeployment):
    """Executes shell scripts from local files"""
    
    def __init__(self, script_file: str, args: List[str] = None, name: str = None,
                 delete: bool = False, timeout: float = None) -> None

Uploads and executes a shell script from a local file.

Parameters:

script_file: Path to local script file
args: Optional command line arguments
name: Optional remote script filename
delete: Whether to delete script after execution
timeout: Optional execution timeout in seconds

Example:

from libcloud.compute.deployment import ScriptFileDeployment

# Execute local installation script
install_deploy = ScriptFileDeployment(
    script_file="/local/scripts/install.sh",
    args=["--verbose", "--config=/etc/app.conf"],
    delete=True,
    timeout=300
)

Multi-Step Deployment

from libcloud.compute.deployment import MultiStepDeployment

class MultiStepDeployment(Deployment):
    """Chains multiple deployment steps together"""
    
    def __init__(self, add: Union[Deployment, List[Deployment]] = None) -> None
    def add(self, add: Union[Deployment, List[Deployment]]) -> None
    def run(self, node: Node, client: BaseSSHClient) -> Node
    
    steps: List[Deployment]

Executes multiple deployment steps in sequence.

Parameters:

add: Initial deployment step(s) to add

Example:

from libcloud.compute.deployment import (
    SSHKeyDeployment, FileDeployment, ScriptDeployment, MultiStepDeployment
)

# Create individual steps
key_step = SSHKeyDeployment(public_key)
config_step = FileDeployment("/local/nginx.conf", "/etc/nginx/nginx.conf")
install_step = ScriptDeployment("apt-get update && apt-get install -y nginx")

# Combine into multi-step deployment
deploy = MultiStepDeployment([key_step, config_step, install_step])

# Or build incrementally
deploy = MultiStepDeployment()
deploy.add(key_step)
deploy.add(config_step)
deploy.add(install_step)

Usage Examples

Basic Node Management

from libcloud.compute.types import Provider
from libcloud.compute.providers import get_driver

# Initialize driver
cls = get_driver(Provider.EC2)
driver = cls('access_key', 'secret_key', region='us-east-1')

# List existing nodes
nodes = driver.list_nodes()
for node in nodes:
    print(f"Node: {node.name}, State: {node.state}, IPs: {node.public_ips}")

# Get available resources
sizes = driver.list_sizes()
images = driver.list_images()
locations = driver.list_locations()

print(f"Available sizes: {len(sizes)}")
print(f"Available images: {len(images)}")
print(f"Available locations: {len(locations)}")

Creating and Managing Nodes

from libcloud.compute.base import NodeAuthSSHKey

# Create SSH key authentication
ssh_key = NodeAuthSSHKey(open('~/.ssh/id_rsa.pub').read())

# Create a new node
node = driver.create_node(
    name='web-server-01',
    size=sizes[0],  # t2.micro equivalent
    image=images[0],  # Ubuntu 20.04 LTS
    location=locations[0],  # us-east-1a
    auth=ssh_key,
    ex_security_groups=['web-servers'],
    ex_userdata='#!/bin/bash\napt-get update'
)

print(f"Created node: {node.name} ({node.id})")

# Wait for node to be running
nodes_running = driver.wait_until_running([node])
print(f"Node is now running: {nodes_running[0][0].state}")

# Reboot the node
success = driver.reboot_node(node)
print(f"Reboot initiated: {success}")

# Destroy the node when done
success = driver.destroy_node(node)
print(f"Node destroyed: {success}")

Working with Multiple Providers

from libcloud.compute.types import Provider
from libcloud.compute.providers import get_driver

# Configure multiple providers
providers_config = {
    'aws': {
        'driver': get_driver(Provider.EC2),
        'credentials': ('aws_access_key', 'aws_secret_key'),
        'region': 'us-east-1'
    },
    'gce': {
        'driver': get_driver(Provider.GCE),
        'credentials': ('service_account_email', 'key_file_path'),
        'project': 'my-project'
    }
}

# Initialize drivers for each provider
drivers = {}
for name, config in providers_config.items():
    cls = config['driver']
    if name == 'aws':
        drivers[name] = cls(*config['credentials'], region=config['region'])
    elif name == 'gce':
        drivers[name] = cls(*config['credentials'], project=config['project'])

# List nodes across all providers
all_nodes = {}
for provider_name, driver in drivers.items():
    nodes = driver.list_nodes()
    all_nodes[provider_name] = nodes
    print(f"{provider_name}: {len(nodes)} nodes")

Advanced Deployment with Custom Scripts

from libcloud.compute.deployment import (
    SSHKeyDeployment, ScriptDeployment, FileDeployment, MultiStepDeployment
)

# Create deployment steps
key_deploy = SSHKeyDeployment(public_key_content)

nginx_config = '''server {
    listen 80;
    location / {
        root /var/www/html;
        index index.html;
    }
}'''

config_deploy = FileDeployment("/local/nginx.conf", "/etc/nginx/sites-available/default")

install_script = ScriptDeployment('''#!/bin/bash
apt-get update
apt-get install -y nginx
systemctl start nginx
systemctl enable nginx
echo "Hello from Libcloud!" > /var/www/html/index.html
systemctl reload nginx
''', timeout=300)

# Combine all steps
multi_deploy = MultiStepDeployment([key_deploy, config_deploy, install_script])

# Deploy node with multi-step deployment
deployed_node = driver.deploy_node(
    name='nginx-server',
    size=sizes[1],
    image=images[0],
    deploy=multi_deploy,
    ssh_username='ubuntu',
    timeout=600
)

print(f"Deployed node with nginx: {deployed_node.public_ips[0]}")
print(f"Exit status: {install_script.exit_status}")

Volume Management

# Create a storage volume
volume = driver.create_volume(
    size=10,  # 10 GB
    name='data-volume',
    location=locations[0]
)
print(f"Created volume: {volume.name} ({volume.id})")

# Attach volume to a node
success = driver.attach_volume(node, volume, device='/dev/sdf')
print(f"Volume attached: {success}")

# Create a snapshot
snapshot = driver.create_volume_snapshot(volume, name='backup-snapshot')
print(f"Created snapshot: {snapshot.id}")

# Detach and cleanup
driver.detach_volume(volume)
driver.destroy_volume_snapshot(snapshot)
driver.destroy_volume(volume)

Exception Handling

from libcloud.common.types import InvalidCredsError, LibcloudError
from libcloud.compute.types import NodeState

try:
    # Attempt operations that may fail
    driver = cls('invalid_key', 'invalid_secret')
    nodes = driver.list_nodes()
except InvalidCredsError:
    print("Invalid credentials provided")
except LibcloudError as e:
    print(f"Libcloud error: {e}")

# Check node state before operations
if node.state == NodeState.RUNNING:
    success = driver.reboot_node(node)
elif node.state == NodeState.STOPPED:
    print("Node is stopped, cannot reboot")

Provider-Specific Features

Many providers offer additional features through the ex_* parameter pattern:

# AWS EC2 specific features
node = driver.create_node(
    name='aws-specific-node',
    size=size,
    image=image,
    ex_security_groups=['web', 'ssh'],
    ex_subnet_id='subnet-12345',
    ex_iam_profile='ec2-role',
    ex_ebs_optimized=True,
    ex_monitoring=True
)

# Google Compute Engine specific features  
node = driver.create_node(
    name='gce-specific-node',
    size=size,
    image=image,
    ex_network='custom-network',
    ex_tags=['web-server', 'production'],
    ex_preemptible=True,
    ex_disk_type='pd-ssd'
)

Check provider-specific documentation and the driver's ex_* methods for additional capabilities.

Install with Tessl CLI