homelab-infra/apps/netbox/terraform/gen-imports.py

#!/usr/bin/env python3
"""Generate imports.tf so Terraform ADOPTS the objects already in NetBox.

    uv run --with requests --with pyyaml python netbox/terraform/gen-imports.py

Run once, when converting an already-populated NetBox to Terraform management. Without it
the first `terraform apply` tries to CREATE objects that exist and fails on uniqueness.

This is a one-shot bootstrap, not part of the normal loop: once `terraform apply` has run,
state holds the IDs and imports.tf can be deleted. Same intent as
../../../infrastructure/openbao/terraform/imports.tf.

It reads the same topology.yml Terraform does, so the resource addresses and for_each keys
line up by construction rather than by hand-transcription.
"""
from __future__ import annotations

import os
import sys
import urllib.parse
from pathlib import Path

import requests
import yaml

HERE = Path(__file__).parent
for envfile in (HERE / "../seed/.env", HERE / ".env"):
    if envfile.exists():
        for line in envfile.read_text().splitlines():
            if line.strip() and not line.startswith("#") and "=" in line:
                k, v = line.split("=", 1)
                os.environ.setdefault(k.strip(), v.strip())

URL = os.environ.get("NETBOX_URL", "https://netbox.ad.ddupan.top").rstrip("/")
TOKEN = os.environ.get("NETBOX_TOKEN", "")
if not TOKEN:
    sys.exit("NETBOX_TOKEN not set")

S = requests.Session()
S.headers.update({"Authorization": f"Bearer {TOKEN}"})


def one(path: str, **params) -> str | None:
    r = S.get(f"{URL}/api{path}", params={**params, "limit": 1}, timeout=30)
    r.raise_for_status()
    res = r.json()["results"]
    return str(res[0]["id"]) if res else None


def main() -> int:
    t = yaml.safe_load((HERE / "topology.yml").read_text())
    out: list[str] = [
        "# GENERATED by gen-imports.py — one-shot bootstrap, safe to delete after the first",
        "# successful `terraform apply`. Do not hand-edit.",
        "",
    ]
    missing: list[str] = []

    def emit(addr: str, oid: str | None, what: str):
        if oid:
            out.append(f'import {{\n  to = {addr}\n  id = "{oid}"\n}}\n')
        else:
            missing.append(what)

    emit("netbox_site.this", one("/dcim/sites/", slug=t["site"]["slug"]), "site")
    emit("netbox_vlan_group.this", one("/ipam/vlan-groups/", slug=t["vlan_group"]["slug"]), "vlan group")

    for r in t["prefix_roles"]:
        emit(f'netbox_ipam_role.this["{r["slug"]}"]', one("/ipam/roles/", slug=r["slug"]), r["slug"])
    for v in t["vlans"]:
        emit(f'netbox_vlan.this["{v["vid"]}"]', one("/ipam/vlans/", vid=v["vid"]), f"vlan {v['vid']}")
    for p in t["prefixes"]:
        emit(f'netbox_prefix.this["{p["prefix"]}"]', one("/ipam/prefixes/", prefix=p["prefix"]), p["prefix"])
    for r in t["ip_ranges"]:
        emit(f'netbox_ip_range.this["{r["start"]}-{r["end"]}"]',
             one("/ipam/ip-ranges/", start_address=r["start"], end_address=r["end"]), "ip range")
    for w in t["wireless_lans"]:
        emit(f'netbox_wireless_lan.this["{w["ssid"]}"]', one("/wireless/wireless-lans/", ssid=w["ssid"]), w["ssid"])
    for m in t["manufacturers"]:
        emit(f'netbox_manufacturer.this["{m["slug"]}"]', one("/dcim/manufacturers/", slug=m["slug"]), m["slug"])
    for d in t["device_types"]:
        emit(f'netbox_device_type.this["{d["slug"]}"]', one("/dcim/device-types/", slug=d["slug"]), d["slug"])
    for r in t["device_roles"]:
        emit(f'netbox_device_role.this["{r["slug"]}"]', one("/dcim/device-roles/", slug=r["slug"]), r["slug"])
    for c in t["cluster_types"]:
        emit(f'netbox_cluster_type.this["{c["slug"]}"]', one("/virtualization/cluster-types/", slug=c["slug"]), c["slug"])
    for c in t["clusters"]:
        emit(f'netbox_cluster.this["{c["name"]}"]', one("/virtualization/clusters/", name=c["name"]), c["name"])

    for d in t["devices"]:
        did = one("/dcim/devices/", name=d["name"])
        emit(f'netbox_device.this["{d["name"]}"]', did, d["name"])
        for i in d["interfaces"]:
            key = f'{d["name"]}:{i["name"]}'
            iid = one("/dcim/interfaces/", device_id=did, name=i["name"]) if did else None
            emit(f'netbox_device_interface.this["{key}"]', iid, key)
            if i.get("ip"):
                emit(f'netbox_ip_address.device["{key}"]', one("/ipam/ip-addresses/", address=i["ip"]), i["ip"])
            # netbox_device_primary_ip is deliberately NOT imported: the provider has no
            # importable object at the device ID ("no object exists with the given id").
            # Letting Terraform "create" it simply re-PATCHes primary_ip4 to the value it
            # already holds, which NetBox treats as a no-op.
            if i.get("mac"):
                mac = i["mac"].upper()
                mid = one("/dcim/mac-addresses/", mac_address=mac)
                emit(f'netbox_mac_address.this["{key}"]', mid, mac)
                if iid:
                    emit(f'netbox_device_interface_primary_mac_address.this["{key}"]', iid, f"{key} primary mac")
        for it in d.get("inventory_items", []):
            key = f'{d["name"]}:{it["name"]}'
            emit(f'netbox_inventory_item.this["{key}"]',
                 one("/dcim/inventory-items/", device_id=did, name=it["name"]) if did else None, key)

    for v in t["virtual_machines"]:
        vid = one("/virtualization/virtual-machines/", name=v["name"])
        emit(f'netbox_virtual_machine.this["{v["name"]}"]', vid, v["name"])
        for i in v["interfaces"]:
            key = f'{v["name"]}:{i["name"]}'
            iid = one("/virtualization/interfaces/", virtual_machine_id=vid, name=i["name"]) if vid else None
            emit(f'netbox_interface.this["{key}"]', iid, key)
            if i.get("ip"):
                emit(f'netbox_ip_address.vm["{key}"]', one("/ipam/ip-addresses/", address=i["ip"]), i["ip"])

    (HERE / "imports.tf").write_text("\n".join(out))
    n = sum(1 for line in out if line.startswith("import {"))
    print(f"wrote imports.tf with {n} import blocks")
    if missing:
        print(f"NOT FOUND in NetBox (Terraform will create these): {', '.join(missing)}")
    return 0


if __name__ == "__main__":
    sys.exit(main())