Python: Discovery – Inventory VMware Hosts and VMs via pyVmomi

export VCENTER_HOST="vcenter.corp.example"
export VCENTER_USER="svc-discovery@vsphere.local"
export VCENTER_PASSWORD="<service-account-password>"

python3 vmware_inventory.py

python3 vmware_inventory.py --host vcenter.lab.example --user svc-discovery@vsphere.local --insecure --output-prefix lab-inventory

Password for svc-discovery@vsphere.local@vcenter.lab.example:
Wrote 4 row(s) to lab-inventory-hosts.csv.
Wrote 37 row(s) to lab-inventory-vms.csv.

vm_name,template,power_state,guest_os,guest_hostname,ip_address,host,cpu_count,memory_gb,committed_gb,provisioned_gb,tools_running,tools_version_status,instance_uuid
app01,False,poweredOn,Microsoft Windows Server 2022 (64-bit),app01.corp.example,10.20.30.41,esx01.lab.example,4,16.0,92.4,120.0,guestToolsRunning,guestToolsCurrent,5003a1c2-...
tpl-rhel9,True,poweredOff,Red Hat Enterprise Linux 9 (64-bit),,,esx02.lab.example,2,4.0,11.2,40.0,guestToolsNotRunning,guestToolsCurrent,5003b7e9-...

"""
Inventory VMware ESXi hosts and virtual machines through the vSphere API (pyVmomi).

Connects to a vCenter Server and uses the PropertyCollector (RetrievePropertiesEx over a
ContainerView) to fetch only the properties it needs, one paged query per object type.
Writes one CSV row per host and one per VM. Intended as a lightweight discovery pass
ahead of a migration, capacity review, or license audit.

Reads: vCenter connection details from CLI arguments or environment variables.
Writes: <output-prefix>-hosts.csv and <output-prefix>-vms.csv.
Requires: Python 3.10+, pyvmomi, a vCenter account with the Read-only role.
"""

import argparse
import csv
import getpass
import os
import sys

from pyVim.connect import Disconnect, SmartConnect
from pyVmomi import vim, vmodl

HOST_PROPERTIES = [
    "name",
    "parent",
    "runtime.connectionState",
    "runtime.inMaintenanceMode",
    "summary.config.product.version",
    "summary.config.product.build",
    "summary.hardware.vendor",
    "summary.hardware.model",
    "summary.hardware.cpuModel",
    "summary.hardware.numCpuPkgs",
    "summary.hardware.numCpuCores",
    "summary.hardware.memorySize",
    "vm",
]

VM_PROPERTIES = [
    "name",
    "summary.config.template",
    "summary.config.guestFullName",
    "summary.config.numCpu",
    "summary.config.memorySizeMB",
    "summary.config.instanceUuid",
    "summary.runtime.powerState",
    "summary.runtime.host",
    "summary.storage.committed",
    "summary.storage.uncommitted",
    "summary.guest.hostName",
    "summary.guest.ipAddress",
    "summary.guest.toolsRunningStatus",
    "summary.guest.toolsVersionStatus2",
]

GIB = 1024 ** 3


def parse_args():
    parser = argparse.ArgumentParser(description="Inventory VMware hosts and VMs via pyVmomi.")
    parser.add_argument("--host", default=os.environ.get("VCENTER_HOST"), help="vCenter Server hostname or IP.")
    parser.add_argument("--user", default=os.environ.get("VCENTER_USER"), help="vCenter user name.")
    parser.add_argument("--password", default=os.environ.get("VCENTER_PASSWORD"), help="vCenter password (prompted if omitted).")
    parser.add_argument("--port", type=int, default=443, help="vCenter API port. Defaults to 443.")
    parser.add_argument("--insecure", action="store_true", help="Skip TLS certificate verification (self-signed vCenter certs).")
    parser.add_argument("--output-prefix", default="vmware-inventory", help="Prefix for the output CSV files.")
    return parser.parse_args()


def connect(args):
    password = args.password or getpass.getpass(f"Password for {args.user}@{args.host}: ")
    return SmartConnect(
        host=args.host,
        user=args.user,
        pwd=password,
        port=args.port,
        disableSslCertValidation=args.insecure,
    )


def collect(content, obj_type, path_set, page_size=500):
    """Yield (managed_object, {property_path: value}) for every obj_type in the inventory."""
    view = content.viewManager.CreateContainerView(content.rootFolder, [obj_type], True)
    try:
        traversal = vmodl.query.PropertyCollector.TraversalSpec(
            name="traverseView", path="view", skip=False, type=vim.view.ContainerView
        )
        obj_spec = vmodl.query.PropertyCollector.ObjectSpec(obj=view, skip=True, selectSet=[traversal])
        prop_spec = vmodl.query.PropertyCollector.PropertySpec(type=obj_type, pathSet=path_set, all=False)
        filter_spec = vmodl.query.PropertyCollector.FilterSpec(objectSet=[obj_spec], propSet=[prop_spec])
        options = vmodl.query.PropertyCollector.RetrieveOptions(maxObjects=page_size)

        collector = content.propertyCollector
        result = collector.RetrievePropertiesEx([filter_spec], options)
        while result:
            for obj in result.objects:
                # Unset properties are simply absent from propSet, so callers use .get().
                yield obj.obj, {prop.name: prop.val for prop in obj.propSet}
            if not result.token:
                break
            result = collector.ContinueRetrievePropertiesEx(result.token)
    finally:
        view.Destroy()


def to_gib(value):
    return round(value / GIB, 1) if value is not None else ""


def inventory_hosts(content):
    rows = []
    host_names = {}
    for host, props in collect(content, vim.HostSystem, HOST_PROPERTIES):
        host_names[host._moId] = props.get("name", "")
        parent = props.get("parent")
        # A clustered host's parent is a ClusterComputeResource; a standalone host
        # sits under a plain ComputeResource that carries the host's own name.
        cluster = parent.name if isinstance(parent, vim.ClusterComputeResource) else "<standalone>"

        rows.append({
            "host_name": props.get("name", ""),
            "cluster": cluster,
            "connection_state": props.get("runtime.connectionState", ""),
            "maintenance_mode": props.get("runtime.inMaintenanceMode", ""),
            "esxi_version": props.get("summary.config.product.version", ""),
            "esxi_build": props.get("summary.config.product.build", ""),
            "vendor": props.get("summary.hardware.vendor", ""),
            "model": props.get("summary.hardware.model", ""),
            "cpu_model": (props.get("summary.hardware.cpuModel") or "").strip(),
            "cpu_sockets": props.get("summary.hardware.numCpuPkgs", ""),
            "cpu_cores": props.get("summary.hardware.numCpuCores", ""),
            "memory_gb": to_gib(props.get("summary.hardware.memorySize")),
            "vm_count": len(props.get("vm", [])),
        })
    return rows, host_names


def inventory_vms(content, host_names):
    rows = []
    for _, props in collect(content, vim.VirtualMachine, VM_PROPERTIES):
        host_ref = props.get("summary.runtime.host")
        committed = props.get("summary.storage.committed")
        uncommitted = props.get("summary.storage.uncommitted")
        memory_mb = props.get("summary.config.memorySizeMB")

        rows.append({
            "vm_name": props.get("name", ""),
            "template": props.get("summary.config.template", ""),
            "power_state": props.get("summary.runtime.powerState", ""),
            "guest_os": props.get("summary.config.guestFullName", ""),
            "guest_hostname": props.get("summary.guest.hostName", ""),
            "ip_address": props.get("summary.guest.ipAddress", ""),
            "host": host_names.get(host_ref._moId, "") if host_ref else "",
            "cpu_count": props.get("summary.config.numCpu", ""),
            "memory_gb": round(memory_mb / 1024, 1) if memory_mb is not None else "",
            "committed_gb": to_gib(committed),
            "provisioned_gb": to_gib((committed or 0) + (uncommitted or 0)) if committed is not None else "",
            "tools_running": props.get("summary.guest.toolsRunningStatus", ""),
            "tools_version_status": props.get("summary.guest.toolsVersionStatus2", ""),
            "instance_uuid": props.get("summary.config.instanceUuid", ""),
        })
    return rows


def write_csv(path, rows):
    if not rows:
        print(f"No rows to write for {path}.")
        return
    with open(path, "w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(rows[0].keys()))
        writer.writeheader()
        writer.writerows(rows)
    print(f"Wrote {len(rows)} row(s) to {path}.")


def main():
    args = parse_args()
    if not args.host or not args.user:
        print("Both --host and --user are required (or VCENTER_HOST / VCENTER_USER).", file=sys.stderr)
        sys.exit(1)

    service_instance = connect(args)
    try:
        content = service_instance.RetrieveContent()
        host_rows, host_names = inventory_hosts(content)
        write_csv(f"{args.output_prefix}-hosts.csv", sorted(host_rows, key=lambda r: r["host_name"]))
        vm_rows = inventory_vms(content, host_names)
        write_csv(f"{args.output_prefix}-vms.csv", sorted(vm_rows, key=lambda r: r["vm_name"]))
    finally:
        Disconnect(service_instance)


if __name__ == "__main__":
    main()