Refactor: Change mixin classes to use Protocol (aristanetworks#4938)

pyproject.toml

@@ -27,7 +27,6 @@ ignore = [
   "TD002",   # We don't have require authors in TODO
   "TD003",   # We don't have an issue link for all TODOs today
   "FIX002",  # Line contains TODO - ignoring for ruff for now
-  "F821",    # Disable undefined-name until resolution of #10451
   "SLF001",  # Accessing private members - TODO: Improve code
   "D100",    # Missing docstring in public module - TODO: Improve code
   "D101",    # Missing docstring in public class - TODO: Improve code

python-avd/pyavd/_cv/client/__init__.py

@@ -4,7 +4,7 @@
 from __future__ import annotations
 
 import ssl
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Protocol
 
 from grpclib.client import Channel
 from requests import JSONDecodeError, get, post
@@ -26,16 +26,19 @@
     from typing_extensions import Self
 
 
-class CVClient(
+class CVClientProtocol(
     ChangeControlMixin,
     ConfigletMixin,
     InventoryMixin,
     StudioMixin,
     SwgMixin,
     TagMixin,
-    UtilsMixin,
     WorkspaceMixin,
+    UtilsMixin,
+    Protocol,
 ):
+    """Protocol for the CVClient class."""
+
     _channel: Channel | None = None
     _metadata: dict
     _servers: list[str]
@@ -46,40 +49,6 @@ class CVClient(
     _password: str | None
     _cv_version: CvVersion | None = None
 
-    def __init__(
-        self,
-        servers: str | list[str],
-        token: str | None = None,
-        username: str | None = None,
-        password: str | None = None,
-        port: int = 443,
-        verify_certs: bool = True,
-    ) -> None:
-        """
-        CVClient is a high-level API library for using CloudVision Resource APIs.
-
-        Use CVClient as an async context manager like:
-            `async with CVClient(servers="myserver", token="mytoken") as cv_client:`
-
-        Parameters:
-            servers: A single FQDN for CVaaS or a list of FQDNs for one CVP cluster.
-            token: Token defined in CloudVision under service-accounts.
-            username: Username to use for authentication if token is not set.
-            password: Password to use for authentication if token is not set.
-            port: TCP port to use for the connection.
-            verify_certs: Disables SSL certificate verification if set to False. Not recommended for production.
-        """
-        if isinstance(servers, list):
-            self._servers = servers
-        else:
-            self._servers = [servers]
-
-        self._port = port
-        self._token = token
-        self._username = username
-        self._password = password
-        self._verify_certs = verify_certs
-
     async def __aenter__(self) -> Self:
         """Using asynchronous context manager since grpclib must be initialized inside an asyncio loop."""
         self._connect()
@@ -176,3 +145,39 @@ def _set_version(self) -> None:
         except (KeyError, JSONDecodeError) as e:
             msg = f"Unable to get version from CloudVision server. Got {response.text}"
             raise CVClientException(msg) from e
+
+
+class CVClient(CVClientProtocol):
+    def __init__(
+        self,
+        servers: str | list[str],
+        token: str | None = None,
+        username: str | None = None,
+        password: str | None = None,
+        port: int = 443,
+        verify_certs: bool = True,
+    ) -> None:
+        """
+        CVClient is a high-level API library for using CloudVision Resource APIs.
+
+        Use CVClient as an async context manager like:
+            `async with CVClient(servers="myserver", token="mytoken") as cv_client:`
+
+        Parameters:
+            servers: A single FQDN for CVaaS or a list of FQDNs for one CVP cluster.
+            token: Token defined in CloudVision under service-accounts.
+            username: Username to use for authentication if token is not set.
+            password: Password to use for authentication if token is not set.
+            port: TCP port to use for the connection.
+            verify_certs: Disables SSL certificate verification if set to False. Not recommended for production.
+        """
+        if isinstance(servers, list):
+            self._servers = servers
+        else:
+            self._servers = [servers]
+
+        self._port = port
+        self._token = token
+        self._username = username
+        self._password = password
+        self._verify_certs = verify_certs

python-avd/pyavd/_cv/client/change_control.py

@@ -4,7 +4,7 @@
 from __future__ import annotations
 
 from logging import getLogger
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING, Literal, Protocol
 
 from pyavd._cv.api.arista.changecontrol.v1 import (
     ApproveConfig,
@@ -32,7 +32,8 @@
 
     from aristaproto import _DateTime
 
-    from . import CVClient
+    from . import CVClientProtocol
+
 
 LOGGER = getLogger(__name__)
 
@@ -44,13 +45,13 @@
 }
 
 
-class ChangeControlMixin:
+class ChangeControlMixin(Protocol):
     """Only to be used as mixin on CVClient class."""
 
     workspace_api_version: Literal["v1"] = "v1"
 
     async def get_change_control(
-        self: CVClient,
+        self: CVClientProtocol,
         change_control_id: str,
         time: datetime | None = None,
         timeout: float = DEFAULT_API_TIMEOUT,
@@ -80,7 +81,7 @@ async def get_change_control(
         return response.value
 
     async def set_change_control(
-        self: CVClient,
+        self: CVClientProtocol,
         change_control_id: str,
         name: str | None = None,
         description: str | None = None,
@@ -115,7 +116,7 @@ async def set_change_control(
         return response.value
 
     async def approve_change_control(
-        self: CVClient,
+        self: CVClientProtocol,
         change_control_id: str,
         timestamp: _DateTime,
         description: str | None = None,
@@ -152,7 +153,7 @@ async def approve_change_control(
         return response.value
 
     async def start_change_control(
-        self: CVClient,
+        self: CVClientProtocol,
         change_control_id: str,
         description: str | None = None,
         timeout: float = DEFAULT_API_TIMEOUT,
@@ -184,7 +185,7 @@ async def start_change_control(
         return response.value
 
     async def wait_for_change_control_state(
-        self: CVClient,
+        self: CVClientProtocol,
         cc_id: str,
         state: Literal["completed", "unspecified", "running", "scheduled"],
         timeout: float = 3600.0,

python-avd/pyavd/_cv/client/configlet.py

@@ -6,7 +6,7 @@
 from asyncio import gather
 from logging import getLogger
 from pathlib import Path
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING, Literal, Protocol
 
 from pyavd._cv.api.arista.configlet.v1 import (
     Configlet,
@@ -38,7 +38,8 @@
 if TYPE_CHECKING:
     from datetime import datetime
 
-    from . import CVClient
+    from . import CVClientProtocol
+
 
 ASSIGNMENT_MATCH_POLICY_MAP = {
     "match_first": MatchPolicy.MATCH_FIRST,
@@ -50,13 +51,13 @@
 LOGGER = getLogger(__name__)
 
 
-class ConfigletMixin:
+class ConfigletMixin(Protocol):
     """Only to be used as mixin on CVClient class."""
 
     configlet_api_version: Literal["v1"] = "v1"
 
     async def get_configlet_containers(
-        self: CVClient,
+        self: CVClientProtocol,
         workspace_id: str,
         container_ids: list[str] | None = None,
         time: datetime | None = None,
@@ -93,7 +94,7 @@ async def get_configlet_containers(
         return configlet_assignments
 
     async def set_configlet_container(
-        self: CVClient,
+        self: CVClientProtocol,
         workspace_id: str,
         container_id: str,
         display_name: str | None = None,
@@ -139,7 +140,7 @@ async def set_configlet_container(
     @LimitCvVersion(min_ver="2024.2.0")
     @grpc_msg_size_handler("containers")
     async def set_configlet_containers(
-        self: CVClient,
+        self: CVClientProtocol,
         workspace_id: str,
         containers: list[tuple[str, str | None, str | None, list[str] | None, str | None, list[str] | None, str | None]],
         timeout: float = DEFAULT_API_TIMEOUT,
@@ -182,7 +183,7 @@ async def set_configlet_containers(
     # Use this variant for versions below 2024.2.0 (still respecting overall min version)
     @LimitCvVersion(max_ver="2024.1.99")
     async def set_configlet_containers(  # noqa: F811 - Redefining with decorator.
-        self: CVClient,
+        self: CVClientProtocol,
         workspace_id: str,
         containers: list[tuple[str, str | None, str | None, list[str] | None, str | None, list[str] | None, str | None]],
         timeout: float = DEFAULT_API_TIMEOUT,
@@ -229,7 +230,7 @@ async def set_configlet_containers(  # noqa: F811 - Redefining with decorator.
         ]
 
     async def delete_configlet_container(
-        self: CVClient,
+        self: CVClientProtocol,
         workspace_id: str,
         assignment_id: str,
         timeout: float = DEFAULT_API_TIMEOUT,
@@ -260,7 +261,7 @@ async def delete_configlet_container(
 
     @grpc_msg_size_handler("configlet_ids")
     async def get_configlets(
-        self: CVClient,
+        self: CVClientProtocol,
         workspace_id: str,
         configlet_ids: list[str] | None = None,
         time: datetime | None = None,
@@ -298,7 +299,7 @@ async def get_configlets(
         return configlets
 
     async def set_configlet(
-        self: CVClient,
+        self: CVClientProtocol,
         workspace_id: str,
         configlet_id: str,
         display_name: str | None = None,
@@ -337,7 +338,7 @@ async def set_configlet(
         return response.value
 
     async def set_configlet_from_file(
-        self: CVClient,
+        self: CVClientProtocol,
         workspace_id: str,
         configlet_id: str,
         file: str,
@@ -378,7 +379,7 @@ async def set_configlet_from_file(
     @LimitCvVersion(min_ver="2024.2.0")
     @grpc_msg_size_handler("configlets")
     async def set_configlets_from_files(
-        self: CVClient,
+        self: CVClientProtocol,
         workspace_id: str,
         configlets: list[tuple[str, str]],
         timeout: float = DEFAULT_API_TIMEOUT,
@@ -419,7 +420,7 @@ async def set_configlets_from_files(
     # Use this variant for versions below 2024.2.0 (still respecting overall min version)
     @LimitCvVersion(max_ver="2024.1.99")
     async def set_configlets_from_files(  # noqa: F811 - Redefining with decorator.
-        self: CVClient,
+        self: CVClientProtocol,
         workspace_id: str,
         configlets: list[tuple[str, str]],
         timeout: float = DEFAULT_API_TIMEOUT,
@@ -456,7 +457,7 @@ async def set_configlets_from_files(  # noqa: F811 - Redefining with decorator.
 
     @grpc_msg_size_handler("configlet_ids")
     async def delete_configlets(
-        self: CVClient,
+        self: CVClientProtocol,
         workspace_id: str,
         configlet_ids: list[str],
         timeout: float = DEFAULT_API_TIMEOUT,

python-avd/pyavd/_cv/client/inventory.py

@@ -3,7 +3,7 @@
 # that can be found in the LICENSE file.
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING, Literal, Protocol
 
 from pyavd._cv.api.arista.inventory.v1 import Device, DeviceKey, DeviceServiceStub, DeviceStreamRequest
 from pyavd._cv.api.arista.time import TimeBounds
@@ -14,16 +14,16 @@
 if TYPE_CHECKING:
     from datetime import datetime
 
-    from . import CVClient
+    from . import CVClientProtocol
 
 
-class InventoryMixin:
+class InventoryMixin(Protocol):
     """Only to be used as mixin on CVClient class."""
 
     inventory_api_version: Literal["v1"] = "v1"
 
     async def get_inventory_devices(
-        self: CVClient,
+        self: CVClientProtocol,
         devices: list[tuple[str, str, str]] | None = None,
         time: datetime | None = None,
         timeout: float = DEFAULT_API_TIMEOUT,