Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Refactor(eos_designs): Adjust class mixin layout and type hints #4922

Draft
wants to merge 4 commits into
base: devel
Choose a base branch
from
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -106,3 +106,13 @@ known-first-party = ["pyavd", "schema_tools"]

[tool.ruff.format]
docstring-code-format = true

[tool.pyright]
include = [
".github",
"python-avd/pyavd/_eos_designs",
]
exclude = [
"python-avd/pyavd/_eos_designs/schema/__init__.py",
]
pythonVersion = "3.10"
5 changes: 2 additions & 3 deletions python-avd/pyavd/_eos_designs/eos_designs_facts/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,13 +10,12 @@

from .mlag import MlagMixin
from .overlay import OverlayMixin
from .short_esi import ShortEsiMixin
from .uplinks import UplinksMixin
from .vlans import VlansMixin
from .wan import WanMixin


class EosDesignsFacts(AvdFacts, MlagMixin, ShortEsiMixin, OverlayMixin, WanMixin, UplinksMixin, VlansMixin):
class EosDesignsFacts(MlagMixin, OverlayMixin, WanMixin, UplinksMixin, VlansMixin, AvdFacts):
"""
`EosDesignsFacts` is based on `AvdFacts`, so make sure to read the description there first.

Expand Down Expand Up @@ -85,7 +84,7 @@ def evpn_multicast(self) -> bool | None:
raise AristaAvdError(msg)

if self.shared_utils.mlag is True:
peer_eos_designs_facts: EosDesignsFacts = self.shared_utils.mlag_peer_facts
peer_eos_designs_facts = self.shared_utils.mlag_peer_facts_cls
if self.shared_utils.overlay_rd_type_admin_subfield == peer_eos_designs_facts.shared_utils.overlay_rd_type_admin_subfield:
msg = "For MLAG devices Route Distinguisher must be unique when 'evpn_multicast: True' since it will create a multi-vtep configuration."
raise AristaAvdError(msg)
Expand Down
18 changes: 8 additions & 10 deletions python-avd/pyavd/_eos_designs/eos_designs_facts/mlag.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,13 +4,11 @@
from __future__ import annotations

from functools import cached_property
from typing import TYPE_CHECKING

if TYPE_CHECKING:
from . import EosDesignsFacts
from .utils import UtilsMixin


class MlagMixin:
class MlagMixin(UtilsMixin):
"""
Mixin Class used to generate some of the EosDesignsFacts.

Expand All @@ -19,35 +17,35 @@ class MlagMixin:
"""

@cached_property
def mlag_peer(self: EosDesignsFacts) -> str | None:
def mlag_peer(self) -> str | None:
"""Exposed in avd_switch_facts."""
if self.shared_utils.mlag:
return self.shared_utils.mlag_peer
return None

@cached_property
def mlag_port_channel_id(self: EosDesignsFacts) -> int | None:
def mlag_port_channel_id(self) -> int | None:
"""Exposed in avd_switch_facts."""
if self.shared_utils.mlag:
return self.shared_utils.mlag_port_channel_id
return None

@cached_property
def mlag_interfaces(self: EosDesignsFacts) -> list | None:
def mlag_interfaces(self) -> list | None:
"""Exposed in avd_switch_facts."""
if self.shared_utils.mlag:
return self.shared_utils.mlag_interfaces
return None

@cached_property
def mlag_ip(self: EosDesignsFacts) -> str | None:
def mlag_ip(self) -> str | None:
"""Exposed in avd_switch_facts."""
if self.shared_utils.mlag:
return self.shared_utils.mlag_ip
return None

@cached_property
def mlag_l3_ip(self: EosDesignsFacts) -> str | None:
def mlag_l3_ip(self) -> str | None:
"""
Exposed in avd_switch_facts.

Expand All @@ -62,7 +60,7 @@ def mlag_l3_ip(self: EosDesignsFacts) -> str | None:
return None

@cached_property
def mlag_switch_ids(self: EosDesignsFacts) -> dict | None:
def mlag_switch_ids(self) -> dict | None:
"""
Exposed in avd_switch_facts.

Expand Down
18 changes: 8 additions & 10 deletions python-avd/pyavd/_eos_designs/eos_designs_facts/overlay.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,13 +4,11 @@
from __future__ import annotations

from functools import cached_property
from typing import TYPE_CHECKING

if TYPE_CHECKING:
from . import EosDesignsFacts
from .utils import UtilsMixin


class OverlayMixin:
class OverlayMixin(UtilsMixin):
"""
Mixin Class used to generate some of the EosDesignsFacts.

Expand All @@ -19,17 +17,17 @@ class OverlayMixin:
"""

@cached_property
def evpn_role(self: EosDesignsFacts) -> str | None:
def evpn_role(self) -> str | None:
"""Exposed in avd_switch_facts."""
return self.shared_utils.evpn_role

@cached_property
def mpls_overlay_role(self: EosDesignsFacts) -> str | None:
def mpls_overlay_role(self) -> str | None:
"""Exposed in avd_switch_facts."""
return self.shared_utils.mpls_overlay_role

@cached_property
def evpn_route_servers(self: EosDesignsFacts) -> list:
def evpn_route_servers(self) -> list:
"""
Exposed in avd_switch_facts.

Expand All @@ -43,7 +41,7 @@ def evpn_route_servers(self: EosDesignsFacts) -> list:
return []

@cached_property
def mpls_route_reflectors(self: EosDesignsFacts) -> list | None:
def mpls_route_reflectors(self) -> list | None:
"""Exposed in avd_switch_facts."""
if self.shared_utils.underlay_router is True and (
self.mpls_overlay_role in ["client", "server"] or (self.evpn_role in ["client", "server"] and self.overlay["evpn_mpls"])
Expand All @@ -52,7 +50,7 @@ def mpls_route_reflectors(self: EosDesignsFacts) -> list | None:
return None

@cached_property
def overlay(self: EosDesignsFacts) -> dict | None:
def overlay(self) -> dict | None:
"""Exposed in avd_switch_facts."""
if self.shared_utils.underlay_router is True:
return {
Expand All @@ -62,7 +60,7 @@ def overlay(self: EosDesignsFacts) -> dict | None:
return None

@cached_property
def vtep_ip(self: EosDesignsFacts) -> str | None:
def vtep_ip(self) -> str | None:
"""Exposed in avd_switch_facts."""
if self.shared_utils.vtep:
return self.shared_utils.vtep_ip
Expand Down
44 changes: 0 additions & 44 deletions python-avd/pyavd/_eos_designs/eos_designs_facts/short_esi.py

This file was deleted.

48 changes: 23 additions & 25 deletions python-avd/pyavd/_eos_designs/eos_designs_facts/uplinks.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,17 +5,15 @@

import re
from functools import cached_property
from typing import TYPE_CHECKING

from pyavd._errors import AristaAvdError
from pyavd._utils import append_if_not_duplicate
from pyavd.j2filters import list_compress, natural_sort, range_expand

if TYPE_CHECKING:
from . import EosDesignsFacts
from .utils import UtilsMixin


class UplinksMixin:
class UplinksMixin(UtilsMixin):
"""
Mixin Class used to generate some of the EosDesignsFacts.

Expand All @@ -24,17 +22,17 @@ class UplinksMixin:
"""

@cached_property
def max_parallel_uplinks(self: EosDesignsFacts) -> int:
def max_parallel_uplinks(self) -> int:
"""Exposed in avd_switch_facts."""
return self.shared_utils.node_config.max_parallel_uplinks

@cached_property
def max_uplink_switches(self: EosDesignsFacts) -> int:
def max_uplink_switches(self) -> int:
"""Exposed in avd_switch_facts."""
return self.shared_utils.max_uplink_switches

@cached_property
def _uplink_port_channel_id(self: EosDesignsFacts) -> int:
def _uplink_port_channel_id(self) -> int:
"""
For MLAG secondary get the uplink_port_channel_id from the peer's facts.

Expand All @@ -49,7 +47,7 @@ def _uplink_port_channel_id(self: EosDesignsFacts) -> int:

if self.shared_utils.mlag_role == "secondary":
# MLAG Secondary
peer_uplink_port_channel_id = self.shared_utils.mlag_peer_facts._uplink_port_channel_id
peer_uplink_port_channel_id = self.shared_utils.mlag_peer_facts_cls._uplink_port_channel_id
# check that port-channel IDs are the same as on primary
if uplink_port_channel_id is not None and uplink_port_channel_id != peer_uplink_port_channel_id:
msg = (
Expand All @@ -72,7 +70,7 @@ def _uplink_port_channel_id(self: EosDesignsFacts) -> int:
return uplink_port_channel_id

@cached_property
def _uplink_switch_port_channel_id(self: EosDesignsFacts) -> int:
def _uplink_switch_port_channel_id(self) -> int:
"""
For MLAG secondary get the uplink_switch_port_channel_id from the peer's facts.

Expand All @@ -88,7 +86,7 @@ def _uplink_switch_port_channel_id(self: EosDesignsFacts) -> int:

if self.shared_utils.mlag_role == "secondary":
# MLAG Secondary
peer_uplink_switch_port_channel_id = self.shared_utils.mlag_peer_facts._uplink_switch_port_channel_id
peer_uplink_switch_port_channel_id = self.shared_utils.mlag_peer_facts_cls._uplink_switch_port_channel_id
# check that port-channel IDs are the same as on primary
if uplink_switch_port_channel_id is not None and uplink_switch_port_channel_id != peer_uplink_switch_port_channel_id:
msg = (
Expand All @@ -104,7 +102,7 @@ def _uplink_switch_port_channel_id(self: EosDesignsFacts) -> int:
uplink_switch_port_channel_id = int("".join(re.findall(r"\d", self.shared_utils.uplink_switch_interfaces[0])))

# produce an error if the uplink switch is MLAG and port-channel ID is above 2000
uplink_switch_facts: EosDesignsFacts = self.shared_utils.get_peer_facts(self.shared_utils.uplink_switches[0], required=True)
uplink_switch_facts = self.shared_utils.get_peer_facts_cls(self.shared_utils.uplink_switches[0])

if uplink_switch_facts.shared_utils.mlag and not 1 <= uplink_switch_port_channel_id <= 2000:
msg = f"'uplink_switch_port_channel_id' must be between 1 and 2000 for MLAG switches. Got '{uplink_switch_port_channel_id}'."
Expand All @@ -113,7 +111,7 @@ def _uplink_switch_port_channel_id(self: EosDesignsFacts) -> int:
return uplink_switch_port_channel_id

@cached_property
def uplinks(self: EosDesignsFacts) -> list:
def uplinks(self) -> list:
"""
Exposed in avd_switch_facts.

Expand Down Expand Up @@ -165,9 +163,9 @@ def uplinks(self: EosDesignsFacts) -> list:

return uplinks

def _get_p2p_uplink(self: EosDesignsFacts, uplink_index: int, uplink_interface: str, uplink_switch: str, uplink_switch_interface: str) -> dict:
def _get_p2p_uplink(self, uplink_index: int, uplink_interface: str, uplink_switch: str, uplink_switch_interface: str) -> dict:
"""Return a single uplink dictionary for uplink_type p2p."""
uplink_switch_facts: EosDesignsFacts = self.shared_utils.get_peer_facts(uplink_switch, required=True)
uplink_switch_facts = self.shared_utils.get_peer_facts_cls(uplink_switch)
uplink = {
"interface": uplink_interface,
"peer": uplink_switch,
Expand Down Expand Up @@ -211,9 +209,9 @@ def _get_p2p_uplink(self: EosDesignsFacts, uplink_index: int, uplink_interface:

return uplink

def _get_port_channel_uplink(self: EosDesignsFacts, uplink_index: int, uplink_interface: str, uplink_switch: str, uplink_switch_interface: str) -> dict:
def _get_port_channel_uplink(self, uplink_index: int, uplink_interface: str, uplink_switch: str, uplink_switch_interface: str) -> dict:
"""Return a single uplink dictionary for uplink_type port-channel."""
uplink_switch_facts: EosDesignsFacts = self.shared_utils.get_peer_facts(uplink_switch, required=True)
uplink_switch_facts = self.shared_utils.get_peer_facts_cls(uplink_switch)

# Reusing get_l2_uplink
uplink = self._get_l2_uplink(uplink_index, uplink_interface, uplink_switch, uplink_switch_interface)
Expand All @@ -229,7 +227,7 @@ def _get_port_channel_uplink(self: EosDesignsFacts, uplink_index: int, uplink_in
uplink["node_group"] = self.shared_utils.group

# Updating unique_uplink_switches with our mlag peer's uplink switches
unique_uplink_switches.update(self.shared_utils.mlag_peer_facts.shared_utils.uplink_switches)
unique_uplink_switches.update(self.shared_utils.mlag_peer_facts_cls.shared_utils.uplink_switches)

# Only enable mlag for this port-channel on the uplink switch if there are multiple unique uplink switches
uplink["peer_mlag"] = len(unique_uplink_switches) > 1
Expand All @@ -240,14 +238,14 @@ def _get_port_channel_uplink(self: EosDesignsFacts, uplink_index: int, uplink_in
return uplink

def _get_l2_uplink(
self: EosDesignsFacts,
self,
uplink_index: int, # pylint: disable=unused-argument # noqa: ARG002
uplink_interface: str,
uplink_switch: str,
uplink_switch_interface: str,
) -> dict:
"""Return a single uplink dictionary for an L2 uplink. Reused for both uplink_type port-channel, lan and TODO: lan-port-channel."""
uplink_switch_facts: EosDesignsFacts = self.shared_utils.get_peer_facts(uplink_switch, required=True)
uplink_switch_facts = self.shared_utils.get_peer_facts_cls(uplink_switch)
uplink = {
"interface": uplink_interface,
"peer": uplink_switch,
Expand Down Expand Up @@ -302,9 +300,9 @@ def _get_l2_uplink(

return uplink

def _get_p2p_vrfs_uplink(self: EosDesignsFacts, uplink_index: int, uplink_interface: str, uplink_switch: str, uplink_switch_interface: str) -> dict:
def _get_p2p_vrfs_uplink(self, uplink_index: int, uplink_interface: str, uplink_switch: str, uplink_switch_interface: str) -> dict:
"""Return a single uplink dictionary for uplink_type p2p-vrfs."""
uplink_switch_facts: EosDesignsFacts = self.shared_utils.get_peer_facts(uplink_switch, required=True)
uplink_switch_facts = self.shared_utils.get_peer_facts_cls(uplink_switch)

# Reusing regular p2p logic for main interface.
uplink = self._get_p2p_uplink(uplink_index, uplink_interface, uplink_switch, uplink_switch_interface)
Expand Down Expand Up @@ -348,7 +346,7 @@ def _get_p2p_vrfs_uplink(self: EosDesignsFacts, uplink_index: int, uplink_interf
return uplink

@cached_property
def uplink_peers(self: EosDesignsFacts) -> list:
def uplink_peers(self) -> list:
"""
Exposed in avd_switch_facts.

Expand All @@ -363,7 +361,7 @@ def uplink_peers(self: EosDesignsFacts) -> list:
return natural_sort(unique_uplink_switches)

@cached_property
def _default_downlink_interfaces(self: EosDesignsFacts) -> list:
def _default_downlink_interfaces(self) -> list:
"""
Internal _default_downlink_interfaces set based on default_interfaces.

Expand All @@ -372,7 +370,7 @@ def _default_downlink_interfaces(self: EosDesignsFacts) -> list:
return range_expand(self.shared_utils.default_interfaces.downlink_interfaces)

@cached_property
def uplink_switch_vrfs(self: EosDesignsFacts) -> list[str] | None:
def uplink_switch_vrfs(self) -> list[str] | None:
"""
Exposed in avd_switch_facts.

Expand All @@ -383,7 +381,7 @@ def uplink_switch_vrfs(self: EosDesignsFacts) -> list[str] | None:

vrfs = set()
for uplink_switch in self.uplink_peers:
uplink_switch_facts = self.shared_utils.get_peer_facts(uplink_switch)
uplink_switch_facts = self.shared_utils.get_peer_facts_cls(uplink_switch)
vrfs.update(uplink_switch_facts.shared_utils.vrfs)

return natural_sort(vrfs) or None
Loading
Loading