From ac763aa37676c645915293b521d59cc390861ff5 Mon Sep 17 00:00:00 2001 From: darnodo Date: Fri, 9 Jan 2026 10:24:57 +0100 Subject: [PATCH] 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 --- pyproject.toml | 2 +- src/netbox/client.py | 136 +++++++++++++++++++++++++++++++++++++------ uv.lock | 2 +- 3 files changed, 120 insertions(+), 20 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index ac4f87b..8ce4d94 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,7 +7,7 @@ requires-python = ">=3.12" dependencies = [ "click>=8.1.0", "pygnmi>=0.8.0", - "pynetbox>=7.0.0", + "pynetbox>=7.5.0", "rich>=13.0.0", ] diff --git a/src/netbox/client.py b/src/netbox/client.py index fceee26..27f1d83 100644 --- a/src/netbox/client.py +++ b/src/netbox/client.py @@ -191,7 +191,10 @@ class FabricNetBoxClient: 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: device_name: Device name @@ -200,14 +203,8 @@ class FabricNetBoxClient: ASN number or None if not assigned """ logger.debug(f"Fetching ASN for device: {device_name}") - try: - asns = self._api.ipam.asns.filter(device=device_name) - 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 + device = self.get_device(device_name) + return device.get("custom_fields", {}).get("asn") # ========================================================================= # Interface Methods @@ -455,23 +452,27 @@ class FabricNetBoxClient: """ 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: - device_name: Optional device name filter + device_name: Ignored - not supported by NetBox API l2vpn_name: Optional L2VPN name filter Returns: List of L2VPN termination dictionaries """ - logger.debug( - f"Fetching L2VPN terminations for device: {device_name or 'all'}, " - f"l2vpn: {l2vpn_name or 'all'}" - ) + if device_name: + logger.warning( + 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: filters = {} if l2vpn_name: filters["l2vpn"] = l2vpn_name - if device_name: - filters["device"] = device_name if filters: terminations = self._api.vpn.l2vpn_terminations.filter(**filters) @@ -547,6 +548,10 @@ class FabricNetBoxClient: """ 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: device_name: Device name vrf_name: VRF name @@ -556,8 +561,25 @@ class FabricNetBoxClient: """ logger.debug(f"Fetching VRF {vrf_name} interfaces for device: {device_name}") try: - interfaces = self._api.dcim.interfaces.filter(device=device_name, vrf=vrf_name) - return [dict(i) for i in interfaces] + # Get IPs in this VRF for this device + 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: raise NetBoxAPIError( 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}" ) 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 # ========================================================================= diff --git a/uv.lock b/uv.lock index 46f38fe..bebdcdf 100644 --- a/uv.lock +++ b/uv.lock @@ -231,7 +231,7 @@ dev = [ requires-dist = [ { name = "click", specifier = ">=8.1.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" }, ]