fix(netbox): adapt client for NetBox 4.4 API compatibility

- Bump pynetbox dependency to >=7.5.0
- Change get_device_asn to read ASN from device custom field instead of
  filtering ASNs by device (unsupported in NetBox 4.4)
- Remove device_name filter from get_l2vpn_terminations as L2VPN
  terminations cannot be filtered by device in NetBox 4.4
- Refactor get_vrf_interfaces to query IP addresses with VRF filter and
  extract associated interfaces (VRF filtering on interfaces unsupported)
- Add get_interface_mlag_id method to read MLAG ID from interface
  custom field
This commit is contained in:
darnodo
2026-01-09 10:24:57 +01:00
parent 2f828f43d6
commit ac763aa376
3 changed files with 120 additions and 20 deletions

View File

@@ -7,7 +7,7 @@ requires-python = ">=3.12"
dependencies = [ dependencies = [
"click>=8.1.0", "click>=8.1.0",
"pygnmi>=0.8.0", "pygnmi>=0.8.0",
"pynetbox>=7.0.0", "pynetbox>=7.5.0",
"rich>=13.0.0", "rich>=13.0.0",
] ]

View File

@@ -191,7 +191,10 @@ class FabricNetBoxClient:
def get_device_asn(self, device_name: str) -> int | None: def get_device_asn(self, device_name: str) -> int | None:
""" """
Get ASN assigned to a device. Get ASN assigned to a device from custom field.
Note: NetBox 4.4 does not support filtering ASNs by device.
ASN is stored as a custom field on the Device object.
Args: Args:
device_name: Device name device_name: Device name
@@ -200,14 +203,8 @@ class FabricNetBoxClient:
ASN number or None if not assigned ASN number or None if not assigned
""" """
logger.debug(f"Fetching ASN for device: {device_name}") logger.debug(f"Fetching ASN for device: {device_name}")
try: device = self.get_device(device_name)
asns = self._api.ipam.asns.filter(device=device_name) return device.get("custom_fields", {}).get("asn")
asn_list = list(asns)
if asn_list:
return asn_list[0].asn
return None
except RequestError as e:
raise NetBoxAPIError(f"Failed to fetch ASN for {device_name}: {e}") from e
# ========================================================================= # =========================================================================
# Interface Methods # Interface Methods
@@ -455,23 +452,27 @@ class FabricNetBoxClient:
""" """
Get L2VPN terminations (VLAN-to-device mappings). Get L2VPN terminations (VLAN-to-device mappings).
Note: NetBox 4.4 does not support filtering L2VPN terminations by device.
L2VPN terminations link to VLANs, not devices directly.
The device_name parameter is accepted for API compatibility but ignored.
Args: Args:
device_name: Optional device name filter device_name: Ignored - not supported by NetBox API
l2vpn_name: Optional L2VPN name filter l2vpn_name: Optional L2VPN name filter
Returns: Returns:
List of L2VPN termination dictionaries List of L2VPN termination dictionaries
""" """
logger.debug( if device_name:
f"Fetching L2VPN terminations for device: {device_name or 'all'}, " logger.warning(
f"l2vpn: {l2vpn_name or 'all'}" f"device_name filter '{device_name}' ignored - "
) "L2VPN terminations cannot be filtered by device in NetBox 4.4"
)
logger.debug(f"Fetching L2VPN terminations for l2vpn: {l2vpn_name or 'all'}")
try: try:
filters = {} filters = {}
if l2vpn_name: if l2vpn_name:
filters["l2vpn"] = l2vpn_name filters["l2vpn"] = l2vpn_name
if device_name:
filters["device"] = device_name
if filters: if filters:
terminations = self._api.vpn.l2vpn_terminations.filter(**filters) terminations = self._api.vpn.l2vpn_terminations.filter(**filters)
@@ -547,6 +548,10 @@ class FabricNetBoxClient:
""" """
Get interfaces assigned to a VRF on a device. Get interfaces assigned to a VRF on a device.
Note: NetBox 4.4 does not support filtering interfaces by VRF directly.
VRF is assigned to IP addresses, not interfaces. This method queries
IP addresses with VRF filter and extracts the associated interfaces.
Args: Args:
device_name: Device name device_name: Device name
vrf_name: VRF name vrf_name: VRF name
@@ -556,8 +561,25 @@ class FabricNetBoxClient:
""" """
logger.debug(f"Fetching VRF {vrf_name} interfaces for device: {device_name}") logger.debug(f"Fetching VRF {vrf_name} interfaces for device: {device_name}")
try: try:
interfaces = self._api.dcim.interfaces.filter(device=device_name, vrf=vrf_name) # Get IPs in this VRF for this device
return [dict(i) for i in interfaces] ips = self._api.ipam.ip_addresses.filter(device=device_name, vrf=vrf_name)
# Extract unique interface IDs
interface_ids = set()
for ip in ips:
if ip.assigned_object_type == "dcim.interface" and ip.assigned_object_id:
interface_ids.add(ip.assigned_object_id)
# Fetch interfaces
if not interface_ids:
return []
interfaces = []
for iid in interface_ids:
iface = self._api.dcim.interfaces.get(id=iid)
if iface:
interfaces.append(dict(iface))
return interfaces
except RequestError as e: except RequestError as e:
raise NetBoxAPIError( raise NetBoxAPIError(
f"Failed to fetch VRF interfaces for {device_name}/{vrf_name}: {e}" f"Failed to fetch VRF interfaces for {device_name}/{vrf_name}: {e}"
@@ -633,6 +655,84 @@ class FabricNetBoxClient:
f"Failed to fetch MLAG peer-link for {device_name}: {e}" f"Failed to fetch MLAG peer-link for {device_name}: {e}"
) from e ) from e
def get_interface_mlag_id(
self, device_name: str, interface_name: str
) -> int | None:
"""
Get MLAG ID from interface custom field.
Args:
device_name: Device name
interface_name: Interface name
Returns:
MLAG ID or None if not assigned
"""
logger.debug(f"Fetching MLAG ID for {device_name}:{interface_name}")
try:
iface = self._api.dcim.interfaces.get(device=device_name, name=interface_name)
if iface:
cf = iface.custom_fields or {}
return cf.get("mlag_id")
return None
except RequestError as e:
raise NetBoxAPIError(
f"Failed to fetch MLAG ID for {device_name}:{interface_name}: {e}"
) from e
# =========================================================================
# VRF Custom Field Methods
# =========================================================================
def get_vrf_l3vni(self, vrf_name: str) -> int | None:
"""
Get L3 VNI from VRF custom field.
Args:
vrf_name: VRF name
Returns:
L3 VNI number or None if not assigned
"""
logger.debug(f"Fetching L3 VNI for VRF: {vrf_name}")
vrf = self.get_vrf(vrf_name)
if vrf:
return vrf.get("custom_fields", {}).get("l3vni")
return None
# =========================================================================
# Virtual IP Methods (Custom Fields)
# =========================================================================
def get_virtual_ips(self, device_name: str | None = None) -> list[dict[str, Any]]:
"""
Get IP addresses marked as virtual/anycast IPs.
Retrieves IPs where the custom field `virtual_ip` is True.
These are typically anycast IPs shared across an MLAG pair.
Args:
device_name: Optional device name filter
Returns:
List of virtual IP address dictionaries
"""
logger.debug(f"Fetching virtual IPs for device: {device_name or 'all'}")
try:
if device_name:
ips = self._api.ipam.ip_addresses.filter(device=device_name)
else:
ips = self._api.ipam.ip_addresses.all()
virtual_ips = []
for ip in ips:
cf = ip.custom_fields or {}
if cf.get("virtual_ip"):
virtual_ips.append(dict(ip))
return virtual_ips
except RequestError as e:
raise NetBoxAPIError(f"Failed to fetch virtual IPs: {e}") from e
# ========================================================================= # =========================================================================
# Aggregated Intent Methods # Aggregated Intent Methods
# ========================================================================= # =========================================================================

2
uv.lock generated
View File

@@ -231,7 +231,7 @@ dev = [
requires-dist = [ requires-dist = [
{ name = "click", specifier = ">=8.1.0" }, { name = "click", specifier = ">=8.1.0" },
{ name = "pygnmi", specifier = ">=0.8.0" }, { name = "pygnmi", specifier = ">=0.8.0" },
{ name = "pynetbox", specifier = ">=7.0.0" }, { name = "pynetbox", specifier = ">=7.5.0" },
{ name = "rich", specifier = ">=13.0.0" }, { name = "rich", specifier = ">=13.0.0" },
] ]