Python SDK

The PiVirt Python SDK provides a native, modern Python client for interacting with PiVirt ConnectRPC and gRPC services, supporting synchronous and asynchronous workflows.

Installation

pip install pivirt
# or
pip install pivirt[grpc]  # with gRPC support

Quick Start

from pivirt.vm.v1 import VMServiceClient

# Create client
client = VMServiceClient(
    'agent.pilab.hu:8443',
    tls_ca='ca.crt',
    tls_cert='client.crt',
    tls_key='client.key',
)

# List VMs
vms = client.list_vms(namespace='production')
for vm in vms:
    print(f"VM: {vm.name} (Status: {vm.status})")

Authentication

mTLS

from pivirt.vm.v1 import VMServiceClient

client = VMServiceClient(
    'agent.pilab.hu:8443',
    tls_ca='ca.crt',
    tls_cert='client.crt',
    tls_key='client.key',
)

Bearer Token

client = VMServiceClient(
    'agent.pilab.hu:8443',
    token='your-jwt-token',
)

VM Operations

Create VM

from pivirt.vm.v1 import VMServiceClient, VMSpec, Disk, NetworkInterface

client = VMServiceClient('agent.pilab.hu:8443')

spec = VMSpec(
    name='web-01',
    vcpus=2,
    memory_bytes=4294967296,
    disks=[
        Disk(
            name='root',
            pool='default',
            size_bytes=21474836480,
            format='QCOW2',
            bus='virtio',
        ),
    ],
    nics=[
        NetworkInterface(
            name='eth0',
            bridge='virbr0',
            model='virtio',
        ),
    ],
)

vm = client.create_vm(spec=spec, namespace='production')
print(f"Created VM: {vm.id}")

Get VM

vm = client.get_vm(id='vm-abc123', namespace='production')
print(f"VM: {vm.name}, Status: {vm.status}")

Start/Stop VM

# Start VM
client.start_vm(id='vm-abc123', namespace='production')

# Stop VM
client.stop_vm(id='vm-abc123', namespace='production')

Migrate VM

operation = client.migrate_vm(
    id='vm-abc123',
    namespace='production',
    target_node='node-02',
    options={
        'live': True,
        'bandwidth_limit_mbps': 1000,
        'downtime_limit_ms': 500,
        'compress': True,
    },
)

print(f"Migration started: {operation.id}")

Storage Operations

Create Volume

from pivirt.storage.v1 import StorageServiceClient

client = StorageServiceClient('agent.pilab.hu:8443')

volume = client.create_volume(
    name='data-01',
    pool='default',
    size_bytes=107374182400,
    format='QCOW2',
)

print(f"Created volume: {volume.id}")

Create Snapshot

snapshot = client.create_snapshot(
    volume_id='vol-abc123',
    name='backup-01',
)

print(f"Created snapshot: {snapshot.id}")

Async Support

import asyncio
from pivirt.vm.v1 import AsyncVMServiceClient

async def main():
    client = AsyncVMServiceClient('agent.pilab.hu:8443')

    vms = await client.list_vms(namespace='production')
    for vm in vms:
        print(f"VM: {vm.name} (Status: {vm.status})")

    await client.close()

asyncio.run(main())

Error Handling

from pivirt.exceptions import PivirtError, NotFoundError, PermissionDeniedError

try:
    vm = client.get_vm(id='vm-abc123')
except NotFoundError:
    print("VM not found")
except PermissionDeniedError:
    print("Permission denied")
except PivirtError as e:
    print(f"Error: {e.message}")

Type Hints

from pivirt.vm.v1.types import VM, VMSpec, VMStatus

def process_vm(vm: VM) -> None:
    print(f"VM: {vm.name}, Status: {vm.status}")

# Type checking with mypy
mypy your_script.py

Examples

See the examples directory for complete examples.