CI: Enable type checking for pyavd._cv [WIP]

.pre-commit-config.yaml

@@ -64,7 +64,7 @@ repos:
           - --allow-past-years
           - --fuzzy-match-generates-todo
           - --comment-style
-          - '{#||#}'
+          - "{#||#}"
           - --no-extra-eol
 
       - id: insert-license
@@ -80,7 +80,7 @@ repos:
           - --detect-license-in-X-top-lines
           - '10'
           - --comment-style
-          - '<!--|  ~|  -->'
+          - "<!--|  ~|  -->"
 
       - name: Check and insert license on Rust files
         id: insert-license
@@ -119,7 +119,8 @@ repos:
           - 'pydantic>=2.3.0'
           - 'jsonschema-rs>=0.24'
           - 'referencing>=0.35.0'
-          - 'ruff==0.11.11'
+          - 'ruff==0.13.0'
+          - "aristaproto>=0.1.1"
 
   - repo: https://github.com/pycqa/pylint
     rev: "v3.3.8"
@@ -162,8 +163,8 @@ repos:
         language: python
         types: [text]
         args:
-          - '--ignore-words=.github/ignore-codespell-words'  # Ignore words listed in this file
-        exclude: >  # List of regex patterns for files/directories to exclude
+          - "--ignore-words=.github/ignore-codespell-words" # Ignore words listed in this file
+        exclude: > # List of regex patterns for files/directories to exclude
           (?x)^(
             .*\.cfg|  # Exclude all .cfg files
             .*\.svg|  # Exclude all .svg files
@@ -180,15 +181,15 @@ repos:
 
   - repo: local
     hooks:
-
       - id: docs-plugin-modules
         name: Build documentation for collection modules and action plugins.
         entry: >-
           ansible-doc-extractor --template docs/templates/plugin-docs.j2
           --markdown "docs/plugins/Modules_and_action_plugins/"
         language: python
         types: [python]
-        additional_dependencies: ['ansible-doc-extractor>=0.1.10', 'ansible-core>=2.16.0,<2.19.0']
+        additional_dependencies:
+          ["ansible-doc-extractor>=0.1.10", "ansible-core>=2.16.0,<2.19.0"]
         files: ansible_collections/arista/avd/plugins/modules/
 
       - id: docs-plugin-filter
@@ -198,7 +199,8 @@ repos:
           --markdown "docs/plugins/Filter_plugins/"
         language: python
         types_or: [python, yaml]
-        additional_dependencies: ['ansible-doc-extractor>=0.1.10', 'ansible-core>=2.16.0,<2.19.0']
+        additional_dependencies:
+          ["ansible-doc-extractor>=0.1.10", "ansible-core>=2.16.0,<2.19.0"]
         files: ansible_collections/arista/avd/plugins/filter/
         exclude: ^(ansible_collections/arista/avd/plugins/filter/deprecated_filters.py)$
 
@@ -209,7 +211,8 @@ repos:
           --markdown "docs/plugins/Lookup_plugins/"
         language: python
         types: [python]
-        additional_dependencies: ['ansible-doc-extractor>=0.1.10', 'ansible-core>=2.16.0,<2.19.0']
+        additional_dependencies:
+          ["ansible-doc-extractor>=0.1.10", "ansible-core>=2.16.0,<2.19.0"]
         files: ansible_collections/arista/avd/plugins/lookup/
 
       - id: docs-plugin-test
@@ -219,7 +222,8 @@ repos:
           --markdown "docs/plugins/Test_plugins/"
         language: python
         types: [python]
-        additional_dependencies: ['ansible-doc-extractor>=0.1.10', 'ansible-core>=2.16.0,<2.19.0']
+        additional_dependencies:
+          ["ansible-doc-extractor>=0.1.10", "ansible-core>=2.16.0,<2.19.0"]
         files: ansible_collections/arista/avd/plugins/test/
 
       - id: docs-plugin-vars
@@ -229,7 +233,8 @@ repos:
           --markdown "docs/plugins/Vars_plugins/"
         language: python
         types: [python]
-        additional_dependencies: ['ansible-doc-extractor>=0.1.10', 'ansible-core>=2.16.0,<2.19.0']
+        additional_dependencies:
+          ["ansible-doc-extractor>=0.1.10", "ansible-core>=2.16.0,<2.19.0"]
         files: ansible_collections/arista/avd/plugins/vars/
 
       - id: schemas
@@ -238,7 +243,15 @@ repos:
         language: python
         files: python-avd/pyavd/[a-z_]+/schema
         pass_filenames: false
-        additional_dependencies: ['deepmerge>=1.1.0', 'PyYAML>=6.0.0', 'pydantic>=2.3.0', 'jsonschema-rs>=0.24', 'referencing>=0.35.0', 'ruff==0.7.2']
+        additional_dependencies:
+          [
+            "deepmerge>=1.1.0",
+            "PyYAML>=6.0.0",
+            "pydantic>=2.3.0",
+            "jsonschema-rs>=0.24",
+            "referencing>=0.35.0",
+            "ruff==0.7.2",
+          ]
 
       - id: check-schema-tables-in-docs
         name: Check for schema tables in documentation.
@@ -265,7 +278,8 @@ repos:
         language: python
         files: python-avd/pyavd/(_eos_cli_config_gen|_eos_designs)/j2templates/
         pass_filenames: false
-        additional_dependencies: ['Jinja2>=3.0.0', 'cryptography>=38.0.4', 'deepmerge>=1.1.0']
+        additional_dependencies:
+          ["Jinja2>=3.0.0", "cryptography>=38.0.4", "deepmerge>=1.1.0"]
 
   - repo: https://github.com/DavidAnson/markdownlint-cli2
     # Keep v0.18.0 for now as pre-commit.ci fails on v0.18.1

pyproject.toml

@@ -125,7 +125,6 @@ exclude = [
   "python/avdutils/tests",                                               # TODO
   "python-avd/pyavd/api",                                                # TODO
   "python-avd/pyavd/_anta",                                              # TODO
-  "python-avd/pyavd/_cv",                                                # TODO
   "python-avd/pyavd/_schema",                                            # TODO
   "python-avd/pyavd/_errors",                                            # TODO
   "python-avd/schema_tools",                                             # TODO

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

@@ -40,7 +40,7 @@ class CVClientProtocol(
     """Protocol for the CVClient class."""
 
     _channel: Channel | None = None
-    _metadata: dict
+    _metadata: dict[str, str]
     _servers: list[str]
     _port: int
     _verify_certs: bool
@@ -59,6 +59,13 @@ async def __aexit__(self, _exc_type: type[BaseException] | None, _exc_val: BaseE
             self._channel.close()
             self._channel = None
 
+    @property
+    def channel(self) -> Channel:
+        if self._channel is None:
+            msg = "'set_change_control' was called with _channel set to None"
+            raise RuntimeError(msg)
+        return self._channel
+
     def _connect(self) -> None:
         # TODO: Verify connection
         # TODO: Handle multinode clusters
@@ -74,7 +81,9 @@ def _connect(self) -> None:
         if self._channel is None:
             self._channel = Channel(host=self._servers[0], port=self._port, ssl=ssl_context)
 
-        self._metadata = {"authorization": "Bearer " + self._token}
+        if self._token is not None:
+            self._metadata = {"authorization": "Bearer " + self._token}
+            # TODO: Should we raise if not token is given
 
     def _ssl_context(self) -> ssl.SSLContext | bool:
         """
@@ -134,14 +143,14 @@ def _set_version(self) -> None:
             msg = "Unable to get version from CloudVision server. Missing token."
             raise CVClientException(msg)
 
-        try:
-            response = get(  # noqa: S113 TODO: Add configurable timeout
-                "https://" + self._servers[0] + "/cvpservice/cvpInfo/getCvpInfo.do",
-                headers={"Authorization": f"Bearer {self._token}"},
-                verify=self._verify_certs,
-                json={},
-            )
+        response = get(  # noqa: S113 TODO: Add configurable timeout
+            "https://" + self._servers[0] + "/cvpservice/cvpInfo/getCvpInfo.do",
+            headers={"Authorization": f"Bearer {self._token}"},
+            verify=self._verify_certs,
+            json={},
+        )
 
+        try:
             self._cv_version = CvVersion(response.json()["version"])
         except (KeyError, JSONDecodeError) as e:
             msg = f"Unable to get version from CloudVision server. Got {response.text}"

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

@@ -24,7 +24,7 @@
 from .versioning import CvVersion
 
 if TYPE_CHECKING:
-    from collections.abc import Callable
+    from collections.abc import Callable, Coroutine
     from inspect import BoundArguments, Signature
 
 LOGGER = getLogger(__name__)
@@ -45,7 +45,7 @@ class LimitCvVersion:
     The decorator will only work in CvClient class methods since it expects the _cv_client attribute on 'self'.
     """
 
-    versioned_funcs: ClassVar[dict[str, dict[tuple[CvVersion, CvVersion], Callable]]] = {}
+    versioned_funcs: ClassVar[dict[str, dict[tuple[CvVersion, CvVersion], Callable[..., Any]]]] = {}
     """
     Map of versioned functions keyed by function name.
 
@@ -70,7 +70,9 @@ def __init__(self, min_ver: str = "2024.1.0", max_ver: str = CVAAS_VERSION_STRIN
             )
             raise ValueError(msg)
 
-    def __call__(self, func: Callable[P, T]) -> Callable[P, T]:
+    # Python 3.12 syntax
+    # def __call__[**P, T](self, func: Callable[P, Coroutine[Any, Any, T]]) -> Callable[P, Coroutine[Any, Any, T]]:
+    def __call__(self, func: Callable[P, Coroutine[Any, Any, T]]) -> Callable[P, Coroutine[Any, Any, T]]:
         """
         Store the method in the map of versioned functions after checking for overlapping decorators for the same method.
 
@@ -130,10 +132,13 @@ class GRPCRequestHandler:
     factor: int
     list_field: str | None
     min_items_for_splitting_attempt: int
-    func: Callable
-    func_signature: Signature
-    bound_arguments: BoundArguments
-    current_arguments_dict: dict
+
+    # These instance variables are set when the decorator is called
+    # Mark them as Optional and initialize to None for Pyright's sake.
+    func: Callable[..., Any] | None = None
+    func_signature: Signature | None = None
+    bound_arguments: BoundArguments | None = None
+    current_arguments_dict: dict[str, Any] | None = None
 
     def __init__(
         self,
@@ -149,7 +154,7 @@ def __init__(
         self.list_field = list_field
         self.min_items_for_splitting_attempt = max(2, min_items_for_splitting_attempt)
 
-    def __call__(self, func: Callable[P, T]) -> Callable[P, T]:
+    def __call__(self, func: Callable[P, Coroutine[Any, Any, T]]) -> Callable[P, Coroutine[Any, Any, T]]:
         self.func = func
         self.func_signature = signature(func)
 
@@ -201,8 +206,11 @@ def _is_list_annotation(annotation: Any, strict: bool = False) -> tuple[bool, An
 
         return _string_based_annotation is list or get_origin(annotation) is list, _string_based_annotation
 
-    async def _execute_single_call_with_retries(self, call_args: tuple, call_kwargs: dict) -> None:
+    async def _execute_single_call_with_retries(self, *call_args: Any, **call_kwargs: Any) -> None:
         """Executes a single call to self.func with retry logic for gRPC UNAVAILABLE."""
+        if self.func is None:
+            # TODO: Add exception to indicate it should not be used directly
+            return None
         func_name = self.func.__name__
 
         for attempt in range(1, self.max_retries + 2):
@@ -243,7 +251,7 @@ async def _execute_single_call_with_retries(self, call_args: tuple, call_kwargs:
                                     raise CVGRPCStatusUnavailable(msg, *e.args, call_args, call_kwargs)
 
                             case Status.RESOURCE_EXHAUSTED:
-                                if matches := fullmatch(MSG_SIZE_EXCEEDED_REGEX, e.message):
+                                if matches := fullmatch(MSG_SIZE_EXCEEDED_REGEX, e.message or ""):
                                     new_exception = CVMessageSizeExceeded(*e.args)
                                     new_exception.max_size = int(matches.group("max"))
                                     new_exception.size = int(matches.group("size"))
@@ -257,7 +265,11 @@ async def _execute_single_call_with_retries(self, call_args: tuple, call_kwargs:
         # Required by ruff
         return None
 
-    async def _execute_with_splitting(self, original_call_args: tuple, original_call_kwargs: dict) -> Any:
+    async def _execute_with_splitting(self, *original_call_args: Any, **original_call_kwargs: Any) -> Any:
+        if self.func is None:
+            # TODO: Add exception to indicate it should not be used directly
+            return None
+
         func_name = self.func.__name__
 
         if not (self.list_field and self.func_signature):
@@ -267,7 +279,7 @@ async def _execute_with_splitting(self, original_call_args: tuple, original_call
         bound_arguments = self.func_signature.bind(*original_call_args, **original_call_kwargs)
         current_arguments_dict = bound_arguments.arguments
 
-        list_value: list = current_arguments_dict.get(self.list_field, [])
+        list_value: list[Any] = current_arguments_dict.get(self.list_field, [])
         if not isinstance(list_value, list):
             msg = (
                 f"{self.__class__.__name__} decorator expected the value of the list_field '{self.list_field}' for function '{func_name}' "

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

@@ -15,7 +15,6 @@
     ChangeControlConfig,
     ChangeControlConfigServiceStub,
     ChangeControlConfigSetRequest,
-    ChangeControlConfigSetResponse,
     ChangeControlKey,
     ChangeControlRequest,
     ChangeControlServiceStub,
@@ -71,9 +70,11 @@ async def get_change_control(
         """
         request = ChangeControlRequest(
             key=ChangeControlKey(id=change_control_id),
-            time=time,
         )
-        client = ChangeControlServiceStub(self._channel)
+        if time is not None:
+            request.time = time
+
+        client = ChangeControlServiceStub(self.channel)
 
         response = await client.get_one(request, metadata=self._metadata, timeout=timeout)
 
@@ -86,7 +87,7 @@ async def set_change_control(
         name: str | None = None,
         description: str | None = None,
         timeout: float = DEFAULT_API_TIMEOUT,
-    ) -> ChangeControlConfigSetResponse:
+    ) -> ChangeControlConfig:
         """
         Set Change Control details using arista.changecontrol.v1.ChangeControlConfigService.Set API.
 
@@ -106,7 +107,7 @@ async def set_change_control(
                 change=ChangeConfig(name=name, notes=description),
             ),
         )
-        client = ChangeControlConfigServiceStub(self._channel)
+        client = ChangeControlConfigServiceStub(self.channel)
 
         response = await client.set(request, metadata=self._metadata, timeout=timeout)
 
@@ -141,7 +142,7 @@ async def approve_change_control(
                 version=timestamp,
             ),
         )
-        client = ApproveConfigServiceStub(self._channel)
+        client = ApproveConfigServiceStub(self.channel)
 
         response = await client.set(request, metadata=self._metadata, timeout=timeout)
 
@@ -171,7 +172,7 @@ async def start_change_control(
                 start=FlagConfig(value=True, notes=description),
             ),
         )
-        client = ChangeControlConfigServiceStub(self._channel)
+        client = ChangeControlConfigServiceStub(self.channel)
 
         response = await client.set(request, metadata=self._metadata, timeout=timeout)
 
@@ -204,7 +205,7 @@ async def wait_for_change_control_state(
                 ),
             ],
         )
-        client = ChangeControlServiceStub(self._channel)
+        client = ChangeControlServiceStub(self.channel)
         responses = client.subscribe(request, metadata=self._metadata, timeout=timeout)
         async for response in responses:
             LOGGER.debug("wait_for_change_control_complete: Response is '%s.'", response)