Refactor(eos_designs): Refactor eos_designs structured_config code for router_path_selection

python-avd/pyavd/_eos_designs/structured_config/network_services/router_path_selection.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 append_if_not_duplicate, get, 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 get
 
 if TYPE_CHECKING:
     from . import AvdStructuredConfigNetworkServicesProtocol
@@ -19,15 +20,13 @@ class RouterPathSelectionMixin(Protocol):
     Class should only be used as Mixin to a AvdStructuredConfig class.
     """
 
-    @cached_property
-    def router_path_selection(self: AvdStructuredConfigNetworkServicesProtocol) -> dict | None:
+    @structured_config_contributor
+    def router_path_selection(self: AvdStructuredConfigNetworkServicesProtocol) -> None:
         """Return structured config for router path-selection (DPS)."""
         if not self.shared_utils.is_wan_router:
-            return None
+            return
 
-        router_path_selection = {
-            "load_balance_policies": self._wan_load_balance_policies(),
-        }
+        self._wan_load_balance_policies()
 
         # When running CV Pathfinder, only load balance policies are configured
         # for AutoVPN, need also vrfs and policies.
@@ -36,53 +35,54 @@ def router_path_selection(self: AvdStructuredConfigNetworkServicesProtocol) -> d
                 {"name": vrf.name, "path_selection_policy": f"{vrf.policy}-WITH-CP" if vrf.name == "default" else vrf.policy} for vrf in self._filtered_wan_vrfs
             ]
 
-            router_path_selection.update(
-                {
-                    "policies": self._autovpn_policies(),
-                    "vrfs": vrfs,
-                },
-            )
+            for vrf in vrfs:
+                self.structured_config.router_path_selection.vrfs.append(EosCliConfigGen.RouterPathSelection.VrfsItem(**vrf))
 
-        return strip_empties_from_dict(router_path_selection)
+            self._autovpn_policies()
 
-    def _wan_load_balance_policies(self: AvdStructuredConfigNetworkServicesProtocol) -> list:
-        """Return a list of load balance policies."""
-        load_balance_policies = []
+    def _wan_load_balance_policies(self: AvdStructuredConfigNetworkServicesProtocol) -> None:
+        """Set list of load balance policies."""
         for policy in self._filtered_wan_policies:
             for match in policy.get("matches", []):
-                append_if_not_duplicate(
-                    list_of_dicts=load_balance_policies,
-                    primary_key="name",
-                    new_dict=match["load_balance_policy"],
-                    context="Router Path-Selection load-balance policies.",
-                    context_keys=["name"],
-                )
-            if (default_match := policy.get("default_match")) is not None:
-                append_if_not_duplicate(
-                    list_of_dicts=load_balance_policies,
-                    primary_key="name",
-                    new_dict=default_match["load_balance_policy"],
-                    context="Router Path-Selection load-balance policies.",
-                    context_keys=["name"],
-                )
+                if "load_balance_policy" in match:
+                    lb_policy = EosCliConfigGen.RouterPathSelection.LoadBalancePoliciesItem(
+                        name=get(match["load_balance_policy"], "name", None),
+                        lowest_hop_count=get(match["load_balance_policy"], "lowest_hop_count", None),
+                        jitter=get(match["load_balance_policy"], "jitter", None),
+                        latency=get(match["load_balance_policy"], "latency", None),
+                        loss_rate=get(match["load_balance_policy"], "loss_rate", None),
+                    )
+                    for group in get(match["load_balance_policy"], "path_groups", None):
+                        path_group_item = EosCliConfigGen.RouterPathSelection.LoadBalancePoliciesItem.PathGroupsItem(
+                            name=group["name"], priority=get(group, "priority", None)
+                        )
+                        lb_policy.path_groups.append(path_group_item)
+                    self.structured_config.router_path_selection.load_balance_policies.append(lb_policy)
 
-        return load_balance_policies
+            if (default_match := policy.get("default_match")) is not None and "load_balance_policy" in default_match:
+                lb_policy = EosCliConfigGen.RouterPathSelection.LoadBalancePoliciesItem(
+                    name=get(default_match["load_balance_policy"], "name", None),
+                )
+                for group in get(default_match["load_balance_policy"], "path_groups", None):
+                    path_group_item = EosCliConfigGen.RouterPathSelection.LoadBalancePoliciesItem.PathGroupsItem(
+                        name=get(group, "name", None), priority=get(group, "priority", None)
+                    )
+                    lb_policy.path_groups.append(path_group_item)
+                self.structured_config.router_path_selection.load_balance_policies.append(lb_policy)
 
-    def _autovpn_policies(self: AvdStructuredConfigNetworkServicesProtocol) -> list:
+    def _autovpn_policies(self: AvdStructuredConfigNetworkServicesProtocol) -> None:
         """Return a list of policies for AutoVPN."""
-        policies = []
         for policy in self._filtered_wan_policies:
-            autovpn_policy = {"name": policy["name"], "rules": []}
+            policy_item = EosCliConfigGen.RouterPathSelection.PoliciesItem()
+            policy_item.name = policy["name"]
             for index, match in enumerate(get(policy, "matches", default=[]), start=1):
-                autovpn_policy["rules"].append(
-                    {
-                        "id": 10 * index,
-                        "application_profile": match["application_profile"],
-                        "load_balance": match["load_balance_policy"]["name"],
-                    },
+                policy_item.rules.append(
+                    EosCliConfigGen.RouterPathSelection.PoliciesItem.RulesItem(
+                        id=10 * index,
+                        application_profile=get(match, "application_profile"),
+                        load_balance=get(match["load_balance_policy"], "name") if "load_balance_policy" in match else None,
+                    )
                 )
-            if (default_match := policy.get("default_match")) is not None:
-                autovpn_policy["default_match"] = {"load_balance": default_match["load_balance_policy"]["name"]}
-
-            policies.append(strip_empties_from_dict(autovpn_policy))
-        return policies
+            if (default_match := policy.get("default_match")) is not None and "load_balance_policy" in default_match:
+                policy_item.default_match.load_balance = get(default_match["load_balance_policy"], "name")
+            self.structured_config.router_path_selection.policies.append(policy_item)