fix(scripts): enhance get_or_create logic and parameter handling

- Replace `endpoint.get()` with `endpoint.filter()` in `get_or_create` to handle object retrieval more robustly and avoid potential exceptions with multiple results.
- Decouple `search_params` from `create_params` to correctly handle differences between API filtering keys (e.g., `group_id`, `vrf_id`) and creation keys (e.g., `group`, `vrf`).
- Refine IP address lookups to include `assigned_object_id` in search parameters, preventing ambiguous matches against IPs not assigned to the target interface.
This commit is contained in:
Damien
2026-02-04 16:31:50 +01:00
parent 52c9586667
commit 0a2f658a2a

View File

@@ -399,11 +399,12 @@ PREFIXES = [
def get_or_create(endpoint, search_params: dict, create_params: dict) -> Any:
"""Get existing object or create new one."""
obj = endpoint.get(**search_params)
if obj:
return obj, False
# Use filter() instead of get() to handle multiple results gracefully
results = list(endpoint.filter(**search_params))
if results:
return results[0], False
obj = endpoint.create({**search_params, **create_params})
obj = endpoint.create(create_params)
return obj, True
@@ -501,6 +502,10 @@ def create_vlans(nb: pynetbox.api, site) -> dict:
result = {}
# VLAN Group
# For NetBox 4.x, scope_type/scope_id are deprecated in favor of scope_object
# but for simple site scope we can just pass scope_type and scope_id or
# rely on backward compatibility if it exists. Ideally we check NetBox version
# but here we'll try to be compatible.
vlan_group, created = get_or_create(
nb.ipam.vlan_groups,
{"slug": VLAN_GROUP_NAME},
@@ -515,7 +520,11 @@ def create_vlans(nb: pynetbox.api, site) -> dict:
vlan_obj, created = get_or_create(
nb.ipam.vlans,
{"vid": vlan["vid"], "group_id": vlan_group.id},
{"name": vlan["name"], "description": vlan.get("description", "")},
{
"name": vlan["name"],
"description": vlan.get("description", ""),
"group": vlan_group.id, # Create expects 'group', filter expects 'group_id'
},
)
log_result("VLAN", "VLAN", f"{vlan['vid']} ({vlan['name']})", created)
result["vlans"][vlan["vid"]] = vlan_obj
@@ -578,10 +587,14 @@ def create_prefixes(nb: pynetbox.api, vrfs: dict):
if vrf_id:
vrf_id = vrf_id.id
create_params = {"prefix": prefix_def["prefix"], "description": prefix_def.get("description", "")}
if vrf_id:
create_params["vrf"] = vrf_id
prefix, created = get_or_create(
nb.ipam.prefixes,
{"prefix": prefix_def["prefix"], "vrf_id": vrf_id},
{"description": prefix_def.get("description", "")},
create_params,
)
log_result("Prefix", "Prefix", prefix_def["prefix"], created)
@@ -736,8 +749,12 @@ def create_ip_addresses(nb: pynetbox.api, devices: dict, interfaces: dict, vrfs:
ip, created = get_or_create(
nb.ipam.ip_addresses,
{"address": spine["loopback0"]},
{
"address": spine["loopback0"],
"assigned_object_id": intf.id,
},
{
"address": spine["loopback0"],
"assigned_object_type": "dcim.interface",
"assigned_object_id": intf.id,
"description": f"{spine['name']} Router-ID",
@@ -753,8 +770,12 @@ def create_ip_addresses(nb: pynetbox.api, devices: dict, interfaces: dict, vrfs:
intf = interfaces[leaf["name"]]["Loopback0"]
ip, created = get_or_create(
nb.ipam.ip_addresses,
{"address": leaf["loopback0"]},
{
"address": leaf["loopback0"],
"assigned_object_id": intf.id,
},
{
"address": leaf["loopback0"],
"assigned_object_type": "dcim.interface",
"assigned_object_id": intf.id,
"description": f"{leaf['name']} Router-ID",
@@ -766,8 +787,12 @@ def create_ip_addresses(nb: pynetbox.api, devices: dict, interfaces: dict, vrfs:
intf = interfaces[leaf["name"]]["Loopback1"]
ip, created = get_or_create(
nb.ipam.ip_addresses,
{"address": leaf["loopback1"]},
{
"address": leaf["loopback1"],
"assigned_object_id": intf.id,
},
{
"address": leaf["loopback1"],
"assigned_object_type": "dcim.interface",
"assigned_object_id": intf.id,
"description": f"{leaf['name']} VTEP",
@@ -781,8 +806,12 @@ def create_ip_addresses(nb: pynetbox.api, devices: dict, interfaces: dict, vrfs:
spine_intf = interfaces["spine1"][f"Ethernet{list(SPINE1_P2P.keys()).index(leaf_name) + 1}"]
ip, created = get_or_create(
nb.ipam.ip_addresses,
{"address": spine_ip},
{
"address": spine_ip,
"assigned_object_id": spine_intf.id,
},
{
"address": spine_ip,
"assigned_object_type": "dcim.interface",
"assigned_object_id": spine_intf.id,
"description": f"spine1 to {leaf_name}",
@@ -794,8 +823,12 @@ def create_ip_addresses(nb: pynetbox.api, devices: dict, interfaces: dict, vrfs:
leaf_intf = interfaces[leaf_name]["Ethernet11"]
ip, created = get_or_create(
nb.ipam.ip_addresses,
{"address": leaf_ip},
{
"address": leaf_ip,
"assigned_object_id": leaf_intf.id,
},
{
"address": leaf_ip,
"assigned_object_type": "dcim.interface",
"assigned_object_id": leaf_intf.id,
"description": f"{leaf_name} to spine1",
@@ -808,8 +841,12 @@ def create_ip_addresses(nb: pynetbox.api, devices: dict, interfaces: dict, vrfs:
spine_intf = interfaces["spine2"][f"Ethernet{list(SPINE2_P2P.keys()).index(leaf_name) + 1}"]
ip, created = get_or_create(
nb.ipam.ip_addresses,
{"address": spine_ip},
{
"address": spine_ip,
"assigned_object_id": spine_intf.id,
},
{
"address": spine_ip,
"assigned_object_type": "dcim.interface",
"assigned_object_id": spine_intf.id,
"description": f"spine2 to {leaf_name}",
@@ -820,8 +857,12 @@ def create_ip_addresses(nb: pynetbox.api, devices: dict, interfaces: dict, vrfs:
leaf_intf = interfaces[leaf_name]["Ethernet12"]
ip, created = get_or_create(
nb.ipam.ip_addresses,
{"address": leaf_ip},
{
"address": leaf_ip,
"assigned_object_id": leaf_intf.id,
},
{
"address": leaf_ip,
"assigned_object_type": "dcim.interface",
"assigned_object_id": leaf_intf.id,
"description": f"{leaf_name} to spine2",