Refactor(eos_designs): structured_config for loopback

python-avd/pyavd/_eos_designs/shared_utils/underlay.py

@@ -6,6 +6,8 @@
 from functools import cached_property
 from typing import TYPE_CHECKING, Protocol
 
+from pyavd._eos_cli_config_gen.schema import EosCliConfigGen
+
 if TYPE_CHECKING:
     from . import SharedUtilsProtocol
 
@@ -60,7 +62,7 @@ def underlay_multicast(self: SharedUtilsProtocol) -> bool:
         return self.inputs.underlay_multicast and self.underlay_router
 
     @cached_property
-    def underlay_multicast_rp_interfaces(self: SharedUtilsProtocol) -> list[dict] | None:
+    def underlay_multicast_rp_interfaces(self: SharedUtilsProtocol) -> list[EosCliConfigGen.LoopbackInterfacesItem] | None:
         if not self.underlay_multicast or not self.inputs.underlay_multicast_rps:
             return None
 
@@ -70,11 +72,11 @@ def underlay_multicast_rp_interfaces(self: SharedUtilsProtocol) -> list[dict] |
                 continue
 
             underlay_multicast_rp_interfaces.append(
-                {
-                    "name": f"Loopback{rp_entry.nodes[self.hostname].loopback_number}",
-                    "description": rp_entry.nodes[self.hostname].description,
-                    "ip_address": f"{rp_entry.rp}/32",
-                },
+                EosCliConfigGen.LoopbackInterfacesItem(
+                    name=f"Loopback{rp_entry.nodes[self.hostname].loopback_number}",
+                    description=rp_entry.nodes[self.hostname].description,
+                    ip_address=f"{rp_entry.rp}/32",
+                )
             )
 
         if underlay_multicast_rp_interfaces:

python-avd/pyavd/_eos_designs/structured_config/network_services/loopback_interfaces.py

@@ -3,10 +3,11 @@
 # that can be found in the LICENSE file.
 from __future__ import annotations
 
-from functools import cached_property
 from typing import TYPE_CHECKING, Protocol
 
-from pyavd._utils import AvdStringFormatter, append_if_not_duplicate, default, strip_empties_from_dict
+from pyavd._eos_cli_config_gen.schema import EosCliConfigGen
+from pyavd._eos_designs.structured_config.structured_config_generator import structured_config_contributor
+from pyavd._utils import AvdStringFormatter, default
 
 if TYPE_CHECKING:
     from pyavd._eos_designs.schema import EosDesigns
@@ -21,67 +22,45 @@ class LoopbackInterfacesMixin(Protocol):
     Class should only be used as Mixin to a AvdStructuredConfig class.
     """
 
-    @cached_property
-    def loopback_interfaces(self: AvdStructuredConfigNetworkServicesProtocol) -> list | None:
+    @structured_config_contributor
+    def loopback_interfaces(self: AvdStructuredConfigNetworkServicesProtocol) -> None:
         """
-        Return structured config for loopback_interfaces.
+        Set the structured config for loopback_interfaces.
 
         Used for Tenant vrf loopback interfaces
         This function is also called from virtual_source_nat_vrfs to avoid duplicate logic
         """
         if not self.shared_utils.network_services_l3:
-            return None
+            return
 
-        loopback_interfaces = []
         for tenant in self.shared_utils.filtered_tenants:
             for vrf in tenant.vrfs:
                 if (loopback_interface := self._get_vtep_diagnostic_loopback_for_vrf(vrf, tenant)) is not None:
-                    append_if_not_duplicate(
-                        list_of_dicts=loopback_interfaces,
-                        primary_key="name",
-                        new_dict=loopback_interface,
-                        context="VTEP Diagnostic Loopback Interfaces",
-                        context_keys=["name", "vrf", "tenant"],
-                        ignore_keys={"tenant"},
-                    )
+                    self._set_virtual_source_nat_for_vrf_loopback(loopback_interface)
+                    self.structured_config.loopback_interfaces.append(loopback_interface)
 
                 # The loopbacks have already been filtered in _filtered_tenants
                 # to only contain entries with our hostname
                 for loopback in vrf.loopbacks:
-                    loopback_interface = {
-                        "name": f"Loopback{loopback.loopback}",
-                        "ip_address": loopback.ip_address,
-                        "shutdown": not loopback.enabled,
-                        "description": loopback.description,
-                        "eos_cli": loopback.raw_eos_cli,
-                    }
-
+                    loopback_interface_item = EosCliConfigGen.LoopbackInterfacesItem(
+                        name=f"Loopback{loopback.loopback}",
+                        ip_address=loopback.ip_address,
+                        shutdown=not loopback.enabled,
+                        description=loopback.description,
+                        eos_cli=loopback.raw_eos_cli,
+                    )
                     if vrf.name != "default":
-                        loopback_interface["vrf"] = vrf.name
-
+                        loopback_interface_item.vrf = vrf.name
                     if loopback.ospf.enabled and vrf.ospf.enabled:
-                        loopback_interface["ospf_area"] = loopback.ospf.area
-
-                    # Strip None values from interface before adding to list
-                    loopback_interface = {key: value for key, value in loopback_interface.items() if value is not None}
-                    append_if_not_duplicate(
-                        list_of_dicts=loopback_interfaces,
-                        primary_key="name",
-                        new_dict=loopback_interface,
-                        context="Loopback Interfaces defined under network_services, vrfs, loopbacks",
-                        context_keys=["name", "vrf"],
-                    )
-
-        if loopback_interfaces:
-            return loopback_interfaces
-
-        return None
+                        loopback_interface_item.ospf_area = loopback.ospf.area
+                    self._set_virtual_source_nat_for_vrf_loopback(loopback_interface_item)
+                    self.structured_config.loopback_interfaces.append(loopback_interface_item)
 
     def _get_vtep_diagnostic_loopback_for_vrf(
         self: AvdStructuredConfigNetworkServicesProtocol,
         vrf: EosDesigns._DynamicKeys.DynamicNetworkServicesItem.NetworkServicesItem.VrfsItem,
         tenant: EosDesigns._DynamicKeys.DynamicNetworkServicesItem.NetworkServicesItem,
-    ) -> dict | None:
+    ) -> EosCliConfigGen.LoopbackInterfacesItem | None:
         if (loopback := vrf.vtep_diagnostic.loopback) is None:
             return None
 
@@ -98,13 +77,11 @@ def _get_vtep_diagnostic_loopback_for_vrf(
 
         interface_name = f"Loopback{loopback}"
         description_template = default(vrf.vtep_diagnostic.loopback_description, self.inputs.default_vrf_diag_loopback_description)
-        return strip_empties_from_dict(
-            {
-                "name": interface_name,
-                "description": AvdStringFormatter().format(description_template, interface=interface_name, vrf=vrf.name, tenant=tenant.name),
-                "shutdown": False,
-                "vrf": vrf.name,
-                "ip_address": f"{self.shared_utils.ip_addressing.vrf_loopback_ip(loopback_ipv4_pool)}/32" if loopback_ipv4_pool else None,
-                "ipv6_address": f"{self.shared_utils.ip_addressing.vrf_loopback_ipv6(loopback_ipv6_pool)}/128" if loopback_ipv6_pool else None,
-            }
+        return EosCliConfigGen.LoopbackInterfacesItem(
+            name=interface_name,
+            description=AvdStringFormatter().format(description_template, interface=interface_name, vrf=vrf.name, tenant=tenant.name),
+            shutdown=False,
+            vrf=vrf.name,
+            ip_address=f"{self.shared_utils.ip_addressing.vrf_loopback_ip(loopback_ipv4_pool)}/32" if loopback_ipv4_pool else None,
+            ipv6_address=f"{self.shared_utils.ip_addressing.vrf_loopback_ipv6(loopback_ipv6_pool)}/128" if loopback_ipv6_pool else None,
         )

python-avd/pyavd/_eos_designs/structured_config/network_services/utils.py

@@ -402,15 +402,15 @@ def get_vrf_router_id(
         # Handle "vtep_diagnostic" router ID case
         if router_id == "diagnostic_loopback":
             # Validate required configuration
-            if (interface_data := self._get_vtep_diagnostic_loopback_for_vrf(vrf, tenant)) is None:
+            if (interface_data := self._get_vtep_diagnostic_loopback_for_vrf(vrf, tenant)) is None or not interface_data.ip_address:
                 msg = (
                     f"Invalid configuration on VRF '{vrf.name}' in Tenant '{tenant.name}'. "
                     "'vtep_diagnostic.loopback' along with either 'vtep_diagnostic.loopback_ip_pools' or 'vtep_diagnostic.loopback_ip_range' must be defined "
                     "when 'router_id' is set to 'diagnostic_loopback' on the VRF."
                 )
                 raise AristaAvdInvalidInputsError(msg)
             # Resolve router ID from loopback interface
-            return get_ip_from_ip_prefix(interface_data["ip_address"])
+            return get_ip_from_ip_prefix(interface_data.ip_address)
         if router_id == "main_router_id":
             return self.shared_utils.router_id if not self.inputs.use_router_general_for_router_id else None
         # Handle "none" router ID

python-avd/pyavd/_eos_designs/structured_config/network_services/virtual_source_nat_vrfs.py

@@ -6,7 +6,6 @@
 from typing import TYPE_CHECKING, Protocol
 
 from pyavd._eos_cli_config_gen.schema import EosCliConfigGen
-from pyavd._eos_designs.structured_config.structured_config_generator import structured_config_contributor
 from pyavd._utils import get_ip_from_ip_prefix
 
 if TYPE_CHECKING:
@@ -20,32 +19,29 @@ class VirtualSourceNatVrfsMixin(Protocol):
     Class should only be used as Mixin to a AvdStructuredConfig class.
     """
 
-    @structured_config_contributor
-    def virtual_source_nat_vrfs(self: AvdStructuredConfigNetworkServicesProtocol) -> None:
+    def _set_virtual_source_nat_for_vrf_loopback(
+        self: AvdStructuredConfigNetworkServicesProtocol, loopback_interface: EosCliConfigGen.LoopbackInterfacesItem
+    ) -> None:
         """
         Set the structured config for virtual_source_nat_vrfs.
 
-        Only used by VTEPs with L2 and L3 services
-        Using data from loopback_interfaces to avoid duplicating logic
+        Only used by VTEPs with L2 and L3 services.
+        Called from loopback_interfaces while creating each loopback.
         """
         if not (self.shared_utils.overlay_vtep and self.shared_utils.network_services_l2 and self.shared_utils.network_services_l3):
             return
 
-        if (loopback_interfaces := self.loopback_interfaces) is None:
+        if (vrf := loopback_interface.vrf) is None:
             return
 
-        for loopback_interface in loopback_interfaces:
-            if (vrf := loopback_interface.get("vrf", "default")) is None:
-                continue
-
-            # Using append with ignore_fields.
-            # It will append the VirtualSourceNatVrfsItem unless the same "name" is already in the list.
-            # It will never raise since we only have these two keys.
-            self.structured_config.virtual_source_nat_vrfs.append(
-                EosCliConfigGen.VirtualSourceNatVrfsItem(
-                    name=vrf,
-                    ip_address=get_ip_from_ip_prefix(loopback_interface["ip_address"]) if "ip_address" in loopback_interface else None,
-                    ipv6_address=get_ip_from_ip_prefix(loopback_interface["ipv6_address"]) if "ipv6_address" in loopback_interface else None,
-                ),
-                ignore_fields=("ip_address", "ipv6_address"),
-            )
+        # Using append with ignore_fields.
+        # It will append the VirtualSourceNatVrfsItem unless the same "name" is already in the list.
+        # It will never raise since we only have these two keys.
+        self.structured_config.virtual_source_nat_vrfs.append(
+            EosCliConfigGen.VirtualSourceNatVrfsItem(
+                name=vrf,
+                ip_address=get_ip_from_ip_prefix(loopback_interface.ip_address) if loopback_interface.ip_address else None,
+                ipv6_address=get_ip_from_ip_prefix(loopback_interface.ipv6_address) if loopback_interface.ipv6_address else None,
+            ),
+            ignore_fields=("ip_address", "ipv6_address"),
+        )

python-avd/pyavd/_eos_designs/structured_config/underlay/loopback_interfaces.py

@@ -6,6 +6,8 @@
 from functools import cached_property
 from typing import TYPE_CHECKING, Protocol
 
+from pyavd._eos_cli_config_gen.schema import EosCliConfigGen
+from pyavd._eos_designs.structured_config.structured_config_generator import structured_config_contributor
 from pyavd._errors import AristaAvdInvalidInputsError
 from pyavd._utils import default
 from pyavd.api.interface_descriptions import InterfaceDescriptionData
@@ -21,84 +23,77 @@ class LoopbackInterfacesMixin(Protocol):
     Class should only be used as Mixin to a AvdStructuredConfig class.
     """
 
-    @cached_property
-    def loopback_interfaces(self: AvdStructuredConfigUnderlayProtocol) -> list | None:
-        """Return structured config for loopback_interfaces."""
+    @structured_config_contributor
+    def loopback_interfaces(self: AvdStructuredConfigUnderlayProtocol) -> None:
+        """Set the structured config for loopback_interfaces."""
         if not self.shared_utils.underlay_router:
-            return None
+            return
 
-        loopback_interfaces = []
         # Loopback 0
-        loopback0 = {
-            "name": "Loopback0",
-            "description": self.shared_utils.interface_descriptions.router_id_loopback_interface(
+        loopback0 = EosCliConfigGen.LoopbackInterfacesItem(
+            name="Loopback0",
+            description=self.shared_utils.interface_descriptions.router_id_loopback_interface(
                 InterfaceDescriptionData(
                     shared_utils=self.shared_utils,
                     interface="Loopback0",
                     description=default(self.inputs.overlay_loopback_description, self.inputs.router_id_loopback_description),
                 ),
             ),
-            "shutdown": False,
-            "ip_address": f"{self.shared_utils.router_id}/32",
-        }
+            shutdown=False,
+            ip_address=f"{self.shared_utils.router_id}/32",
+        )
 
         if self.shared_utils.ipv6_router_id is not None:
-            loopback0["ipv6_address"] = f"{self.shared_utils.ipv6_router_id}/128"
+            loopback0.ipv6_address = f"{self.shared_utils.ipv6_router_id}/128"
 
         if self.shared_utils.underlay_ospf:
-            loopback0["ospf_area"] = self.inputs.underlay_ospf_area
+            loopback0.ospf_area = self.inputs.underlay_ospf_area
 
         if self.shared_utils.underlay_ldp:
-            loopback0["mpls"] = {"ldp": {"interface": True}}
+            loopback0.mpls.ldp.interface = True
 
         if self.shared_utils.underlay_isis:
-            isis_config = {"isis_enable": self.shared_utils.isis_instance_name, "isis_passive": True}
+            loopback0._update(isis_enable=self.shared_utils.isis_instance_name, isis_passive=True)
             if self.shared_utils.underlay_sr:
-                isis_config["node_segment"] = {"ipv4_index": self._node_sid}
+                loopback0.node_segment.ipv4_index = self._node_sid
                 if self.shared_utils.underlay_ipv6:
-                    isis_config["node_segment"].update({"ipv6_index": self._node_sid})
-
-            loopback0.update(isis_config)
-
-        loopback0 = {key: value for key, value in loopback0.items() if value is not None}
+                    loopback0.node_segment.ipv6_index = self._node_sid
 
-        loopback_interfaces.append(loopback0)
+        self.structured_config.loopback_interfaces.append(loopback0)
 
         # VTEP loopback
         if (
             self.shared_utils.overlay_vtep is True
             and self.shared_utils.vtep_loopback.lower() != "loopback0"
             and self.shared_utils.vtep_loopback.lower().startswith("lo")
         ):
-            vtep_loopback = {
-                "name": self.shared_utils.vtep_loopback,
-                "description": self.shared_utils.interface_descriptions.vtep_loopback_interface(
+            vtep_loopback = EosCliConfigGen.LoopbackInterfacesItem(
+                name=self.shared_utils.vtep_loopback,
+                description=self.shared_utils.interface_descriptions.vtep_loopback_interface(
                     InterfaceDescriptionData(
                         shared_utils=self.shared_utils, interface=self.shared_utils.vtep_loopback, description=self.inputs.vtep_loopback_description
                     )
-                ),
-                "shutdown": False,
-                "ip_address": f"{self.shared_utils.vtep_ip}/32",
-            }
+                )
+                or None,
+                shutdown=False,
+                ip_address=f"{self.shared_utils.vtep_ip}/32",
+            )
 
             if self.shared_utils.network_services_l3 is True and self.inputs.vtep_vvtep_ip is not None:
-                vtep_loopback["ip_address_secondaries"] = [self.inputs.vtep_vvtep_ip]
+                vtep_loopback.ip_address_secondaries.append_new(self.inputs.vtep_vvtep_ip)
 
             if self.shared_utils.underlay_ospf is True:
-                vtep_loopback["ospf_area"] = self.inputs.underlay_ospf_area
+                vtep_loopback.ospf_area = self.inputs.underlay_ospf_area
 
             if self.shared_utils.underlay_isis is True:
-                vtep_loopback.update({"isis_enable": self.shared_utils.isis_instance_name, "isis_passive": True})
-
-            vtep_loopback = {key: value for key, value in vtep_loopback.items() if value is not None}
+                vtep_loopback._update(isis_enable=self.shared_utils.isis_instance_name, isis_passive=True)
 
-            loopback_interfaces.append(vtep_loopback)
+            self.structured_config.loopback_interfaces.append(vtep_loopback)
 
         # Underlay Multicast RP Loopbacks
         if self.shared_utils.underlay_multicast_rp_interfaces is not None:
-            loopback_interfaces.extend(self.shared_utils.underlay_multicast_rp_interfaces)
-
-        return loopback_interfaces
+            for underlay_multicast_rp_interface in self.shared_utils.underlay_multicast_rp_interfaces:
+                self.structured_config.loopback_interfaces.append(underlay_multicast_rp_interface)
 
     @cached_property
     def _node_sid(self: AvdStructuredConfigUnderlayProtocol) -> int:

python-avd/pyavd/_eos_designs/structured_config/underlay/prefix_lists.py

@@ -55,7 +55,7 @@ def prefix_lists(self: AvdStructuredConfigUnderlayProtocol) -> list | None:
 
         if self.shared_utils.underlay_multicast_rp_interfaces is not None:
             sequence_numbers = [
-                {"sequence": (index + 1) * 10, "action": f"permit {interface['ip_address']}"}
+                {"sequence": (index + 1) * 10, "action": f"permit {interface.ip_address}"}
                 for index, interface in enumerate(self.shared_utils.underlay_multicast_rp_interfaces)
             ]
             prefix_lists.append({"name": "PL-LOOPBACKS-PIM-RP", "sequence_numbers": sequence_numbers})