feat(anta): Add timestamps option to asynceapi client and update typing

anta/cli/exec/utils.py

@@ -23,6 +23,7 @@
 
 if TYPE_CHECKING:
     from anta.inventory import AntaInventory
+    from asynceapi._types import EapiComplexCommand, EapiSimpleCommand
 
 EOS_SCHEDULED_TECH_SUPPORT = "/mnt/flash/schedule/tech-support"
 INVALID_CHAR = "`~!@#$/"
@@ -135,7 +136,7 @@ async def collect(device: AntaDevice) -> None:
                 )
                 logger.warning(msg)
 
-                commands = []
+                commands: list[EapiSimpleCommand | EapiComplexCommand] = []
                 # TODO: @mtache - add `config` field to `AntaCommand` object to handle this use case.
                 # Otherwise mypy complains about enable as it is only implemented for AsyncEOSDevice
                 # TODO: Should enable be also included in AntaDevice?

anta/device.py

@@ -26,6 +26,8 @@
     from collections.abc import Iterator
     from pathlib import Path
 
+    from asynceapi._types import EapiComplexCommand, EapiSimpleCommand
+
 logger = logging.getLogger(__name__)
 
 # Do not load the default keypairs multiple times due to a performance issue introduced in cryptography 37.0
@@ -445,7 +447,7 @@ async def _collect(self, command: AntaCommand, *, collection_id: str | None = No
         collection_id
             An identifier used to build the eAPI request ID.
         """
-        commands: list[dict[str, str | int]] = []
+        commands: list[EapiComplexCommand | EapiSimpleCommand] = []
         if self.enable and self._enable_password is not None:
             commands.append(
                 {
@@ -458,12 +460,12 @@ async def _collect(self, command: AntaCommand, *, collection_id: str | None = No
             commands.append({"cmd": "enable"})
         commands += [{"cmd": command.command, "revision": command.revision}] if command.revision else [{"cmd": command.command}]
         try:
-            response: list[dict[str, Any] | str] = await self._session.cli(
+            response = await self._session.cli(
                 commands=commands,
                 ofmt=command.ofmt,
                 version=command.version,
                 req_id=f"ANTA-{collection_id}-{id(command)}" if collection_id else f"ANTA-{id(command)}",
-            )  # type: ignore[assignment] # multiple commands returns a list
+            )
             # Do not keep response of 'enable' command
             command.output = response[-1]
         except asynceapi.EapiCommandError as e:

asynceapi/_types.py

@@ -0,0 +1,43 @@
+# Copyright (c) 2024-2025 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""Type definitions used for the asynceapi package."""
+
+from __future__ import annotations
+
+from typing import Any, Literal, NotRequired, TypedDict
+
+EapiJsonOutput = dict[str, Any]
+"""Type definition of an eAPI JSON output response."""
+EapiTextOutput = str
+"""Type definition of an eAPI text output response."""
+EapiSimpleCommand = str
+"""Type definition of an eAPI simple command."""
+
+
+class EapiComplexCommand(TypedDict):
+    """Type definition of an eAPI complex command."""
+
+    cmd: str
+    input: NotRequired[str]
+    revision: NotRequired[int]
+
+
+class JsonRpc(TypedDict):
+    """Type definition of a JSON-RPC payload."""
+
+    jsonrpc: Literal["2.0"]
+    method: Literal["runCmds"]
+    params: JsonRpcParams
+    id: NotRequired[int | str]
+
+
+class JsonRpcParams(TypedDict):
+    """Type definition of JSON-RPC parameters."""
+
+    version: NotRequired[int | Literal["latest"]]
+    cmds: list[EapiSimpleCommand | EapiComplexCommand]
+    format: NotRequired[Literal["json", "text"]]
+    autoComplete: NotRequired[bool]
+    expandAliases: NotRequired[bool]
+    timestamps: NotRequired[bool]

asynceapi/config_session.py

@@ -10,9 +10,10 @@
 from __future__ import annotations
 
 import re
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
+    from ._types import EapiComplexCommand, EapiJsonOutput, EapiSimpleCommand
     from .device import Device
 
 # -----------------------------------------------------------------------------
@@ -78,15 +79,15 @@ def device(self) -> Device:
     #                               Public Methods
     # -------------------------------------------------------------------------
 
-    async def status_all(self) -> dict[str, Any]:
+    async def status_all(self) -> EapiJsonOutput:
         """Get the status of all the session config on the device.
 
         Run the following command on the device:
             # show configuration sessions detail
 
         Returns
         -------
-        dict[str, Any]
+        EapiJsonOutput
             Dictionary of native EOS eAPI response; see `status` method for
             details.
 
@@ -116,9 +117,9 @@ async def status_all(self) -> dict[str, Any]:
         }
         ```
         """
-        return await self._cli("show configuration sessions detail")  # type: ignore[return-value] # json outformat returns dict[str, Any]
+        return await self._cli(command="show configuration sessions detail")
 
-    async def status(self) -> dict[str, Any] | None:
+    async def status(self) -> EapiJsonOutput | None:
         """Get the status of a session config on the device.
 
         Run the following command on the device:
@@ -129,7 +130,7 @@ async def status(self) -> dict[str, Any] | None:
 
         Returns
         -------
-        dict[str, Any] | None
+        EapiJsonOutput | None
             Dictionary instance of the session status. If the session does not exist,
             then this method will return None.
 
@@ -201,7 +202,7 @@ async def push(self, content: list[str] | str, *, replace: bool = False) -> None
         # prepare the initial set of command to enter the config session and
         # rollback clean if the `replace` argument is True.
 
-        commands: list[str | dict[str, Any]] = [self._cli_config_session]
+        commands: list[EapiSimpleCommand | EapiComplexCommand] = [self._cli_config_session]
         if replace:
             commands.append(self.CLI_CFG_FACTORY_RESET)
 
@@ -232,15 +233,15 @@ async def commit(self, timer: str | None = None) -> None:
         if timer:
             command += f" timer {timer}"
 
-        await self._cli(command)
+        await self._cli(command=command)
 
     async def abort(self) -> None:
         """Abort the configuration session.
 
         Run the following command on the device:
             # configure session <name> abort
         """
-        await self._cli(f"{self._cli_config_session} abort")
+        await self._cli(command=f"{self._cli_config_session} abort")
 
     async def diff(self) -> str:
         """Return the "diff" of the session config relative to the running config.
@@ -257,7 +258,7 @@ async def diff(self) -> str:
         ----------
             * https://www.gnu.org/software/diffutils/manual/diffutils.txt
         """
-        return await self._cli(f"show session-config named {self.name} diffs", ofmt="text")  # type: ignore[return-value] # text outformat returns str
+        return await self._cli(command=f"show session-config named {self.name} diffs", ofmt="text")
 
     async def load_file(self, filename: str, *, replace: bool = False) -> None:
         """Load the configuration from <filename> into the session configuration.
@@ -281,12 +282,12 @@ async def load_file(self, filename: str, *, replace: bool = False) -> None:
             If there are any issues with loading the configuration file then a
             RuntimeError is raised with the error messages content.
         """
-        commands: list[str | dict[str, Any]] = [self._cli_config_session]
+        commands: list[EapiSimpleCommand | EapiComplexCommand] = [self._cli_config_session]
         if replace:
             commands.append(self.CLI_CFG_FACTORY_RESET)
 
         commands.append(f"copy {filename} session-config")
-        res: list[dict[str, Any]] = await self._cli(commands=commands)  # type: ignore[assignment] # JSON outformat of multiple commands returns list[dict[str, Any]]
+        res = await self._cli(commands=commands)
         checks_re = re.compile(r"error|abort|invalid", flags=re.IGNORECASE)
         messages = res[-1]["messages"]
 
@@ -295,4 +296,4 @@ async def load_file(self, filename: str, *, replace: bool = False) -> None:
 
     async def write(self) -> None:
         """Save the running config to the startup config by issuing the command "write" to the device."""
-        await self._cli("write")
+        await self._cli(command="write")