Repository: aristanetworks/anta
Pull request #1060 (2025-03-03 15:40:58)

feat(anta): Add EapiRequest and EapiResponse models for asynceapi to support eAPI stopOnError and timestamps features
# Description

Adding internal modules to asynceapi to support request/response models when executing commands on eAPI.

The goal is to have a consistent response model with whatever eAPI JSON-RPC params we use in the requests.

New models now support `timestamps` and `stopOnError` eAPI features properly.

# Checklist:

<!-- Delete not relevant items !-->

- [X] My code follows the style guidelines of this project
- [X] I have performed a self-review of my code
- [X] I have run pre-commit for code linting and typing (`pre-commit run`)
- [ ] I have made corresponding changes to the documentation
- [x] I have added tests that prove my fix is effective or that my feature works
- [x] New and existing unit tests pass locally with my changes (`tox -e testenv`)


sonarqubecloud[bot]: ## [![Quality Gate Failed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-failed-20px.png 'Quality Gate Failed')](https://sonarcloud.io/dashboard?id=aristanetworks_anta&pullRequest=1060) **Quality Gate failed**  
Failed conditions  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/failed-16px.png '') [41.8% Coverage on New Code](https://sonarcloud.io/component_measures?id=aristanetworks_anta&pullRequest=1060&metric=new_coverage&view=list) (required ≥ 80%)  
  
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=aristanetworks_anta&pullRequest=1060)


sonarqubecloud[bot]: ## [![Quality Gate Passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-passed-20px.png 'Quality Gate Passed')](https://sonarcloud.io/dashboard?id=aristanetworks_anta&pullRequest=1060) **Quality Gate passed**  
Issues  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [1 New issue](https://sonarcloud.io/project/issues?id=aristanetworks_anta&pullRequest=1060&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/accepted-16px.png '') [0 Accepted issues](https://sonarcloud.io/project/issues?id=aristanetworks_anta&pullRequest=1060&issueStatuses=ACCEPTED)

Measures  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=aristanetworks_anta&pullRequest=1060&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [97.8% Coverage on New Code](https://sonarcloud.io/component_measures?id=aristanetworks_anta&pullRequest=1060&metric=new_coverage&view=list)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=aristanetworks_anta&pullRequest=1060&metric=new_duplicated_lines_density&view=list)  
  
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=aristanetworks_anta&pullRequest=1060)


codspeed-hq[bot]: <!-- __CODSPEED_PERFORMANCE_REPORT_COMMENT__ -->
<!-- __CODSPEED_INSTRUMENTATION_PERFORMANCE_REPORT_COMMENT__ -->

## [CodSpeed Performance Report](https://codspeed.io/aristanetworks/anta/branches/carl-baillargeon%3Afeat%2Fasynceapi_stop_on_error)

### Merging #1060 will **not alter performance**

<sub>Comparing <code>carl-baillargeon:feat/asynceapi_stop_on_error</code> (319702a) with <code>main</code> (2672570)</sub>



### Summary

`✅ 22` untouched benchmarks  




sonarqubecloud[bot]: ## [![Quality Gate Passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-passed-20px.png 'Quality Gate Passed')](https://sonarcloud.io/dashboard?id=aristanetworks_anta&pullRequest=1060) **Quality Gate passed**  
Issues  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [1 New issue](https://sonarcloud.io/project/issues?id=aristanetworks_anta&pullRequest=1060&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/accepted-16px.png '') [0 Accepted issues](https://sonarcloud.io/project/issues?id=aristanetworks_anta&pullRequest=1060&issueStatuses=ACCEPTED)

Measures  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=aristanetworks_anta&pullRequest=1060&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [97.8% Coverage on New Code](https://sonarcloud.io/component_measures?id=aristanetworks_anta&pullRequest=1060&metric=new_coverage&view=list)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=aristanetworks_anta&pullRequest=1060&metric=new_duplicated_lines_density&view=list)  
  
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=aristanetworks_anta&pullRequest=1060)


sonarqubecloud[bot]: ## [![Quality Gate Passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-passed-20px.png 'Quality Gate Passed')](https://sonarcloud.io/dashboard?id=aristanetworks_anta&pullRequest=1060) **Quality Gate passed**  
Issues  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [1 New issue](https://sonarcloud.io/project/issues?id=aristanetworks_anta&pullRequest=1060&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/accepted-16px.png '') [0 Accepted issues](https://sonarcloud.io/project/issues?id=aristanetworks_anta&pullRequest=1060&issueStatuses=ACCEPTED)

Measures  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=aristanetworks_anta&pullRequest=1060&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [97.7% Coverage on New Code](https://sonarcloud.io/component_measures?id=aristanetworks_anta&pullRequest=1060&metric=new_coverage&view=list)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=aristanetworks_anta&pullRequest=1060&metric=new_duplicated_lines_density&view=list)  
  
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=aristanetworks_anta&pullRequest=1060)



Patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 6e9430db..c3663715 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -98,7 +98,7 @@ repos:
           - types-requests
           - types-pyOpenSSL
           - pytest
-        files: ^(anta|tests)/
+        files: ^(anta|tests|asynceapi)/
 
   - repo: https://github.com/igorshubovych/markdownlint-cli
     rev: v0.44.0
diff --git a/asynceapi/_constants.py b/asynceapi/_constants.py
new file mode 100644
index 00000000..2904038a
--- /dev/null
+++ b/asynceapi/_constants.py
@@ -0,0 +1,22 @@
+# 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.
+"""Constants and Enums for the asynceapi package."""
+
+from __future__ import annotations
+
+from enum import Enum
+
+
+class EapiCommandFormat(str, Enum):
+    """Enum for the eAPI command format.
+
+    NOTE: This could be updated to StrEnum when Python 3.11 is the minimum supported version in ANTA.
+    """
+
+    JSON = "json"
+    TEXT = "text"
+
+    def __str__(self) -> str:
+        """Override the __str__ method to return the value of the Enum, mimicking the behavior of StrEnum."""
+        return self.value
diff --git a/asynceapi/_errors.py b/asynceapi/_errors.py
new file mode 100644
index 00000000..321843db
--- /dev/null
+++ b/asynceapi/_errors.py
@@ -0,0 +1,36 @@
+# 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.
+"""Exceptions for the asynceapi package."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from asynceapi._models import EapiResponse
+
+
+class EapiReponseError(Exception):
+    """Exception raised when an eAPI response contains errors.
+
+    Attributes
+    ----------
+    response : EapiResponse
+        The eAPI response that contains the error.
+    """
+
+    def __init__(self, response: EapiResponse) -> None:
+        """Initialize the EapiReponseError exception."""
+        self.response = response
+
+        # Build a descriptive error message
+        message = "Error in eAPI response"
+
+        if response.error_code is not None:
+            message += f" (code: {response.error_code})"
+
+        if response.error_message is not None:
+            message += f": {response.error_message}"
+
+        super().__init__(message)
diff --git a/asynceapi/_models.py b/asynceapi/_models.py
new file mode 100644
index 00000000..0572a2f3
--- /dev/null
+++ b/asynceapi/_models.py
@@ -0,0 +1,238 @@
+# 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.
+"""Models for the asynceapi package."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from logging import getLogger
+from typing import TYPE_CHECKING, Any, Literal
+from uuid import uuid4
+
+from ._constants import EapiCommandFormat
+from ._errors import EapiReponseError
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+    from ._types import EapiComplexCommand, EapiJsonOutput, EapiSimpleCommand, EapiTextOutput, JsonRpc
+
+LOGGER = getLogger(__name__)
+
+
+# pylint: disable=too-many-instance-attributes
+@dataclass(frozen=True)
+class EapiRequest:
+    """Model for an eAPI request.
+
+    Attributes
+    ----------
+    commands : list[EapiSimpleCommand | EapiComplexCommand]
+        A list of commands to execute.
+    version : int | Literal["latest"]
+        The eAPI version to use. Defaults to "latest".
+    format : EapiCommandFormat
+        The command output format. Defaults "json".
+    timestamps : bool
+        Include timestamps in the command output. Defaults to False.
+    auto_complete : bool
+        Enable command auto-completion. Defaults to False.
+    expand_aliases : bool
+        Expand command aliases. Defaults to False.
+    stop_on_error : bool
+        Stop command execution on first error. Defaults to True.
+    id : int | str
+        The request ID. Defaults to a random hex string.
+    """
+
+    commands: list[EapiSimpleCommand | EapiComplexCommand]
+    version: int | Literal["latest"] = "latest"
+    format: EapiCommandFormat = EapiCommandFormat.JSON
+    timestamps: bool = False
+    auto_complete: bool = False
+    expand_aliases: bool = False
+    stop_on_error: bool = True
+    id: int | str = field(default_factory=lambda: uuid4().hex)
+
+    def to_jsonrpc(self) -> JsonRpc:
+        """Return the JSON-RPC dictionary payload for the request."""
+        return {
+            "jsonrpc": "2.0",
+            "method": "runCmds",
+            "params": {
+                "version": self.version,
+                "cmds": self.commands,
+                "format": self.format,
+                "timestamps": self.timestamps,
+                "autoComplete": self.auto_complete,
+                "expandAliases": self.expand_aliases,
+                "stopOnError": self.stop_on_error,
+            },
+            "id": self.id,
+        }
+
+
+@dataclass(frozen=True)
+class EapiResponse:
+    """Model for an eAPI response.
+
+    Construct an EapiResponse from a JSON-RPC response dictionary using the `from_jsonrpc` class method.
+
+    Can be iterated over to access command results in order of execution.
+
+    Attributes
+    ----------
+    request_id : str
+        The ID of the original request this response corresponds to.
+    _results : dict[int, EapiCommandResult]
+        Dictionary mapping request command indices to their respective results.
+    error_code : int | None
+        The JSON-RPC error code, if any.
+    error_message : str | None
+        The JSON-RPC error message, if any.
+    """
+
+    request_id: str
+    _results: dict[int, EapiCommandResult] = field(default_factory=dict)
+    error_code: int | None = None
+    error_message: str | None = None
+
+    @property
+    def success(self) -> bool:
+        """Return True if the response has no errors."""
+        return self.error_code is None
+
+    @property
+    def results(self) -> list[EapiCommandResult]:
+        """Get all results as a list. Results are ordered by the command indices in the request."""
+        return list(self._results.values())
+
+    def __len__(self) -> int:
+        """Return the number of results."""
+        return len(self._results)
+
+    def __iter__(self) -> Iterator[EapiCommandResult]:
+        """Enable iteration over the results. Results are yielded in the same order as provided in the request."""
+        yield from self._results.values()
+
+    @classmethod
+    def from_jsonrpc(cls, response: dict[str, Any], request: EapiRequest, *, raise_on_error: bool = False) -> EapiResponse:
+        """Build an EapiResponse from a JSON-RPC eAPI response.
+
+        Parameters
+        ----------
+        response
+            The JSON-RPC eAPI response dictionary.
+        request
+            The corresponding EapiRequest.
+        raise_on_error
+            Raise an EapiReponseError if the response contains errors, by default False.
+
+        Returns
+        -------
+        EapiResponse
+            The EapiResponse object.
+        """
+        has_error = "error" in response
+        response_data = response["error"]["data"] if has_error else response["result"]
+
+        # Handle case where we have fewer results than commands (stop_on_error=True)
+        executed_count = min(len(response_data), len(request.commands))
+
+        # Process the results we have
+        results = {}
+        for i in range(executed_count):
+            cmd = request.commands[i]
+            cmd_str = cmd["cmd"] if isinstance(cmd, dict) else cmd
+            data = response_data[i]
+
+            output = None
+            errors = []
+            success = True
+            start_time = None
+            duration = None
+
+            # Parse the output based on the data type, no output when errors are present
+            if isinstance(data, dict):
+                if "errors" in data:
+                    errors = data["errors"]
+                    success = False
+                else:
+                    output = data["output"] if request.format == EapiCommandFormat.TEXT and "output" in data else data
+
+                # Add timestamps if available
+                if request.timestamps and "_meta" in data:
+                    meta = data.pop("_meta")
+                    start_time = meta.get("execStartTime")
+                    duration = meta.get("execDuration")
+
+            elif isinstance(data, str):
+                # Handle case where eAPI returns a JSON string response (serialized JSON) for certain commands
+                try:
+                    from json import JSONDecodeError, loads
+
+                    output = loads(data)
+                except (JSONDecodeError, TypeError):
+                    # If it's not valid JSON, store as is
+                    LOGGER.warning("Invalid JSON response for command: %s. Storing as text: %s", cmd_str, data)
+                    output = data
+
+            results[i] = EapiCommandResult(
+                command=cmd_str,
+                output=output,
+                errors=errors,
+                success=success,
+                start_time=start_time,
+                duration=duration,
+            )
+
+        # If stop_on_error is True and we have an error, indicate commands not executed
+        if has_error and request.stop_on_error and executed_count < len(request.commands):
+            for i in range(executed_count, len(request.commands)):
+                cmd = request.commands[i]
+                cmd_str = cmd["cmd"] if isinstance(cmd, dict) else cmd
+                results[i] = EapiCommandResult(command=cmd_str, output=None, errors=["Command not executed due to previous error"], success=False, executed=False)
+
+        response_obj = cls(
+            request_id=response["id"],
+            _results=results,
+            error_code=response["error"]["code"] if has_error else None,
+            error_message=response["error"]["message"] if has_error else None,
+        )
+
+        if raise_on_error and has_error:
+            raise EapiReponseError(response_obj)
+
+        return response_obj
+
+
+@dataclass(frozen=True)
+class EapiCommandResult:
+    """Model for an eAPI command result.
+
+    Attributes
+    ----------
+    command : str
+        The command that was executed.
+    output : EapiJsonOutput | EapiTextOutput | None
+        The command result output. None if the command returned errors.
+    errors : list[str]
+        A list of error messages, if any.
+    success : bool
+        True if the command was successful.
+    executed : bool
+        True if the command was executed. When `stop_on_error` is True in the request, some commands may not be executed.
+    start_time : float | None
+        Command execution start time in seconds. Uses Unix epoch format. `timestamps` must be True in the request.
+    duration : float | None
+        Command execution duration in seconds. `timestamps` must be True in the request.
+    """
+
+    command: str
+    output: EapiJsonOutput | EapiTextOutput | None
+    errors: list[str] = field(default_factory=list)
+    success: bool = True
+    executed: bool = True
+    start_time: float | None = None
+    duration: float | None = None
diff --git a/asynceapi/_types.py b/asynceapi/_types.py
index 675b7715..ebebf04b 100644
--- a/asynceapi/_types.py
+++ b/asynceapi/_types.py
@@ -6,7 +6,10 @@
 from __future__ import annotations
 
 import sys
-from typing import Any, Literal
+from typing import TYPE_CHECKING, Any, Literal
+
+if TYPE_CHECKING:
+    from ._constants import EapiCommandFormat
 
 if sys.version_info >= (3, 11):
     from typing import NotRequired, TypedDict
@@ -18,11 +21,11 @@ EapiJsonOutput = dict[str, Any]
 EapiTextOutput = str
 """Type definition of an eAPI text output response."""
 EapiSimpleCommand = str
-"""Type definition of an eAPI simple command."""
+"""Type definition of an eAPI simple command. A simple command is the CLI command to run as a string."""
 
 
 class EapiComplexCommand(TypedDict):
-    """Type definition of an eAPI complex command."""
+    """Type definition of an eAPI complex command. A complex command is a dictionary with the CLI command to run with additional parameters."""
 
     cmd: str
     input: NotRequired[str]
@@ -43,7 +46,8 @@ class JsonRpcParams(TypedDict):
 
     version: NotRequired[int | Literal["latest"]]
     cmds: list[EapiSimpleCommand | EapiComplexCommand]
-    format: NotRequired[Literal["json", "text"]]
+    format: NotRequired[EapiCommandFormat]
     autoComplete: NotRequired[bool]
     expandAliases: NotRequired[bool]
     timestamps: NotRequired[bool]
+    stopOnError: NotRequired[bool]
diff --git a/asynceapi/device.py b/asynceapi/device.py
index 907ecb33..a7702daa 100644
--- a/asynceapi/device.py
+++ b/asynceapi/device.py
@@ -20,6 +20,7 @@ import httpx
 # -----------------------------------------------------------------------------
 # Private Imports
 # -----------------------------------------------------------------------------
+from ._constants import EapiCommandFormat
 from .aio_portcheck import port_check_url
 from .config_session import SessionConfig
 from .errors import EapiCommandError
@@ -133,7 +134,6 @@ class Device(httpx.AsyncClient):
         suppress_error: Literal[False] = False,
         auto_complete: bool = False,
         expand_aliases: bool = False,
-        timestamps: bool = False,
         req_id: int | str | None = None,
     ) -> EapiJsonOutput: ...
 
@@ -149,7 +149,6 @@ class Device(httpx.AsyncClient):
         suppress_error: Literal[False] = False,
         auto_complete: bool = False,
         expand_aliases: bool = False,
-        timestamps: bool = False,
         req_id: int | str | None = None,
     ) -> list[EapiJsonOutput]: ...
 
@@ -165,7 +164,6 @@ class Device(httpx.AsyncClient):
         suppress_error: Literal[False] = False,
         auto_complete: bool = False,
         expand_aliases: bool = False,
-        timestamps: bool = False,
         req_id: int | str | None = None,
     ) -> EapiTextOutput: ...
 
@@ -181,7 +179,6 @@ class Device(httpx.AsyncClient):
         suppress_error: Literal[False] = False,
         auto_complete: bool = False,
         expand_aliases: bool = False,
-        timestamps: bool = False,
         req_id: int | str | None = None,
     ) -> list[EapiTextOutput]: ...
 
@@ -197,7 +194,6 @@ class Device(httpx.AsyncClient):
         suppress_error: Literal[True],
         auto_complete: bool = False,
         expand_aliases: bool = False,
-        timestamps: bool = False,
         req_id: int | str | None = None,
     ) -> EapiJsonOutput | None: ...
 
@@ -213,7 +209,6 @@ class Device(httpx.AsyncClient):
         suppress_error: Literal[True],
         auto_complete: bool = False,
         expand_aliases: bool = False,
-        timestamps: bool = False,
         req_id: int | str | None = None,
     ) -> list[EapiJsonOutput] | None: ...
 
@@ -229,7 +224,6 @@ class Device(httpx.AsyncClient):
         suppress_error: Literal[True],
         auto_complete: bool = False,
         expand_aliases: bool = False,
-        timestamps: bool = False,
         req_id: int | str | None = None,
     ) -> EapiTextOutput | None: ...
 
@@ -245,7 +239,6 @@ class Device(httpx.AsyncClient):
         suppress_error: Literal[True],
         auto_complete: bool = False,
         expand_aliases: bool = False,
-        timestamps: bool = False,
         req_id: int | str | None = None,
     ) -> list[EapiTextOutput] | None: ...
 
@@ -260,7 +253,6 @@ class Device(httpx.AsyncClient):
         suppress_error: bool = False,
         auto_complete: bool = False,
         expand_aliases: bool = False,
-        timestamps: bool = False,
         req_id: int | str | None = None,
     ) -> EapiJsonOutput | EapiTextOutput | list[EapiJsonOutput] | list[EapiTextOutput] | None:
         """Execute one or more CLI commands.
@@ -302,8 +294,6 @@ class Device(httpx.AsyncClient):
                 For example if an alias is configured as 'sv' for 'show version'
                 then an API call with sv and the expandAliases parameter will
                 return the output of show version.
-        timestamps
-            If True, return the per-command execution time.
         req_id
             A unique identifier that will be echoed back by the switch. May be a string or number.
 
@@ -336,7 +326,6 @@ class Device(httpx.AsyncClient):
             version=version,
             auto_complete=auto_complete,
             expand_aliases=expand_aliases,
-            timestamps=timestamps,
             req_id=req_id,
         )
 
@@ -356,7 +345,6 @@ class Device(httpx.AsyncClient):
         *,
         auto_complete: bool = False,
         expand_aliases: bool = False,
-        timestamps: bool = False,
         req_id: int | str | None = None,
     ) -> JsonRpc:
         """Create the JSON-RPC command dictionary object.
@@ -387,8 +375,6 @@ class Device(httpx.AsyncClient):
                 For example if an alias is configured as 'sv' for 'show version'
                 then an API call with sv and the expandAliases parameter will
                 return the output of show version.
-        timestamps
-            If True, return the per-command execution time.
         req_id
             A unique identifier that will be echoed back by the switch. May be a string or number.
 
@@ -404,10 +390,9 @@ class Device(httpx.AsyncClient):
             "params": {
                 "version": version,
                 "cmds": commands,
-                "format": ofmt,
+                "format": EapiCommandFormat(ofmt),
                 "autoComplete": auto_complete,
                 "expandAliases": expand_aliases,
-                "timestamps": timestamps,
             },
             "id": req_id or id(self),
         }
diff --git a/pyproject.toml b/pyproject.toml
index cedb9821..176ca0ff 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -271,12 +271,14 @@ commands =
   ruff format . --check
   pylint anta
   pylint tests
+  pylint asynceapi
 
 [testenv:type]
 description = Check typing
 commands =
   mypy --config-file=pyproject.toml anta
   mypy --config-file=pyproject.toml tests
+  mypy --config-file=pyproject.toml asynceapi
 
 [testenv:clean]
 description = Erase previous coverage reports


Test patch:
diff --git a/tests/units/asynceapi/test__constants.py b/tests/units/asynceapi/test__constants.py
new file mode 100644
index 00000000..aee1a484
--- /dev/null
+++ b/tests/units/asynceapi/test__constants.py
@@ -0,0 +1,42 @@
+# Copyright (c) 2023-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.
+"""Unit tests for the asynceapi._constants module."""
+
+import pytest
+
+from asynceapi._constants import EapiCommandFormat
+
+
+class TestEapiCommandFormat:
+    """Test cases for the EapiCommandFormat enum."""
+
+    def test_enum_values(self) -> None:
+        """Test that the enum has the expected values."""
+        assert EapiCommandFormat.JSON.value == "json"
+        assert EapiCommandFormat.TEXT.value == "text"
+
+    def test_str_method(self) -> None:
+        """Test that the __str__ method returns the string value."""
+        assert str(EapiCommandFormat.JSON) == "json"
+        assert str(EapiCommandFormat.TEXT) == "text"
+
+        # Test in string formatting
+        assert f"Format: {EapiCommandFormat.JSON}" == "Format: json"
+
+    def test_string_behavior(self) -> None:
+        """Test that the enum behaves like a string."""
+        # String methods should work
+        assert EapiCommandFormat.JSON.upper() == "JSON"
+
+        # String comparisons should work
+        assert EapiCommandFormat.JSON == "json"
+        assert EapiCommandFormat.TEXT == "text"
+
+    def test_enum_lookup(self) -> None:
+        """Test enum lookup by value."""
+        assert EapiCommandFormat("json") is EapiCommandFormat.JSON
+        assert EapiCommandFormat("text") is EapiCommandFormat.TEXT
+
+        with pytest.raises(ValueError, match="'invalid' is not a valid EapiCommandFormat"):
+            EapiCommandFormat("invalid")
diff --git a/tests/units/asynceapi/test__models.py b/tests/units/asynceapi/test__models.py
new file mode 100644
index 00000000..447c2433
--- /dev/null
+++ b/tests/units/asynceapi/test__models.py
@@ -0,0 +1,435 @@
+# Copyright (c) 2023-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.
+"""Unit tests for the asynceapi._models module."""
+
+import logging
+from typing import TYPE_CHECKING
+from uuid import UUID
+
+import pytest
+
+from asynceapi._constants import EapiCommandFormat
+from asynceapi._errors import EapiReponseError
+from asynceapi._models import EapiCommandResult, EapiRequest, EapiResponse
+
+if TYPE_CHECKING:
+    from asynceapi._types import EapiComplexCommand, EapiSimpleCommand
+
+
+class TestEapiRequest:
+    """Test cases for the EapiRequest class."""
+
+    def test_init_with_defaults(self) -> None:
+        """Test initialization with just required parameters."""
+        commands: list[EapiSimpleCommand | EapiComplexCommand] = ["show version", "show interfaces"]
+        req = EapiRequest(commands=commands)
+
+        # Check required attributes
+        assert req.commands == commands
+
+        # Check default values
+        assert req.version == "latest"
+        assert req.format == EapiCommandFormat.JSON
+        assert req.timestamps is False
+        assert req.auto_complete is False
+        assert req.expand_aliases is False
+        assert req.stop_on_error is True
+
+        # Check that ID is generated as a UUID hex string
+        try:
+            UUID(str(req.id))
+            is_valid_uuid = True
+        except ValueError:
+            is_valid_uuid = False
+        assert is_valid_uuid
+
+    def test_init_with_custom_values(self) -> None:
+        """Test initialization with custom parameter values."""
+        commands: list[EapiSimpleCommand | EapiComplexCommand] = [{"cmd": "enable", "input": "password"}, "show version"]
+        req = EapiRequest(
+            commands=commands,
+            version=1,
+            format=EapiCommandFormat.TEXT,
+            timestamps=True,
+            auto_complete=True,
+            expand_aliases=True,
+            stop_on_error=False,
+            id="custom-id-123",
+        )
+
+        # Check all attributes match expected values
+        assert req.commands == commands
+        assert req.version == 1
+        assert req.format == EapiCommandFormat.TEXT
+        assert req.timestamps is True
+        assert req.auto_complete is True
+        assert req.expand_aliases is True
+        assert req.stop_on_error is False
+        assert req.id == "custom-id-123"
+
+    def test_to_jsonrpc(self) -> None:
+        """Test conversion to JSON-RPC dictionary."""
+        commands: list[EapiSimpleCommand | EapiComplexCommand] = ["show version", "show interfaces"]
+        req = EapiRequest(commands=commands, version=1, format=EapiCommandFormat.TEXT, id="test-id-456")
+
+        jsonrpc = req.to_jsonrpc()
+
+        # Check that structure matches expected JSON-RPC format
+        assert jsonrpc["jsonrpc"] == "2.0"
+        assert jsonrpc["method"] == "runCmds"
+        assert jsonrpc["id"] == "test-id-456"
+
+        # Check params matches our configuration
+        params = jsonrpc["params"]
+        assert params["version"] == 1
+        assert params["cmds"] == commands
+        assert params["format"] == EapiCommandFormat.TEXT
+        assert params["timestamps"] is False
+        assert params["autoComplete"] is False
+        assert params["expandAliases"] is False
+        assert params["stopOnError"] is True
+
+    def test_to_jsonrpc_with_complex_commands(self) -> None:
+        """Test JSON-RPC conversion with complex commands."""
+        commands: list[EapiSimpleCommand | EapiComplexCommand] = [
+            {"cmd": "enable", "input": "password"},
+            {"cmd": "configure", "input": ""},
+            {"cmd": "hostname test-device"},
+        ]
+        req = EapiRequest(commands=commands)
+
+        jsonrpc = req.to_jsonrpc()
+
+        # Verify commands are passed correctly
+        assert jsonrpc["params"]["cmds"] == commands
+
+    def test_immutability(self) -> None:
+        """Test that the dataclass is truly immutable (frozen)."""
+        commands: list[EapiSimpleCommand | EapiComplexCommand] = ["show version"]
+        req = EapiRequest(commands=commands)
+
+        # Attempting to modify any attribute should raise an error
+        with pytest.raises(AttributeError):
+            req.commands = ["new command"]  # type: ignore[misc]
+
+        with pytest.raises(AttributeError):
+            req.id = "new-id"  # type: ignore[misc]
+
+
+class TestEapiResponse:
+    """Test cases for the EapiResponse class."""
+
+    def test_init_and_properties(self) -> None:
+        """Test basic initialization and properties."""
+        # Create mock command results
+        result1 = EapiCommandResult(command="show version", output={"version": "4.33.2F-40713977.4332F (engineering build)"})
+        result2 = EapiCommandResult(command="show hostname", output={"hostname": "DC1-LEAF1A"})
+
+        # Create response with results
+        results = {0: result1, 1: result2}
+        response = EapiResponse(request_id="test-123", _results=results)
+
+        # Check attributes
+        assert response.request_id == "test-123"
+        assert response.error_code is None
+        assert response.error_message is None
+
+        # Check properties
+        assert response.success is True
+        assert len(response.results) == 2
+        assert response.results[0] == result1
+        assert response.results[1] == result2
+
+    def test_error_response(self) -> None:
+        """Test initialization with error information."""
+        result = EapiCommandResult(command="show bad command", output=None, errors=["Invalid input (at token 1: 'bad')"], success=False)
+        results = {0: result}
+        response = EapiResponse(
+            request_id="test-456", _results=results, error_code=1002, error_message="CLI command 1 of 1 'show bad command' failed: invalid command"
+        )
+
+        assert response.request_id == "test-456"
+        assert response.error_code == 1002
+        assert response.error_message == "CLI command 1 of 1 'show bad command' failed: invalid command"
+        assert response.success is False
+        assert len(response.results) == 1
+        assert response.results[0].success is False
+        assert "Invalid input (at token 1: 'bad')" in response.results[0].errors
+
+    def test_len_and_iteration(self) -> None:
+        """Test __len__ and __iter__ methods."""
+        # Create 3 command results
+        results = {
+            0: EapiCommandResult(command="cmd1", output="out1"),
+            1: EapiCommandResult(command="cmd2", output="out2"),
+            2: EapiCommandResult(command="cmd3", output="out3"),
+        }
+        response = EapiResponse(request_id="test-789", _results=results)
+
+        # Test __len__
+        assert len(response) == 3
+
+        # Test __iter__
+        iterated_results = list(response)
+        assert len(iterated_results) == 3
+        assert [r.command for r in iterated_results] == ["cmd1", "cmd2", "cmd3"]
+
+    def test_from_jsonrpc_success(self) -> None:
+        """Test from_jsonrpc with successful response."""
+        # Mock request
+        request = EapiRequest(commands=["show version", "show hostname"], format=EapiCommandFormat.JSON)
+
+        # Mock response data
+        jsonrpc_response = {
+            "jsonrpc": "2.0",
+            "id": "test-id-123",
+            "result": [{"modelName": "cEOSLab", "version": "4.33.2F-40713977.4332F (engineering build)"}, {"hostname": "DC1-LEAF1A"}],
+        }
+
+        response = EapiResponse.from_jsonrpc(jsonrpc_response, request)
+
+        # Verify response object
+        assert response.request_id == "test-id-123"
+        assert response.success is True
+        assert response.error_code is None
+        assert response.error_message is None
+
+        # Verify results
+        assert len(response) == 2
+        assert response.results[0].command == "show version"
+        assert response.results[0].output == {"modelName": "cEOSLab", "version": "4.33.2F-40713977.4332F (engineering build)"}
+        assert response.results[0].success is True
+        assert response.results[1].command == "show hostname"
+        assert response.results[1].output == {"hostname": "DC1-LEAF1A"}
+        assert response.results[1].success is True
+
+    def test_from_jsonrpc_text_format(self) -> None:
+        """Test from_jsonrpc with TEXT format responses."""
+        # Mock request with TEXT format
+        request = EapiRequest(commands=["show version", "show hostname"], format=EapiCommandFormat.TEXT)
+
+        # Mock response data
+        jsonrpc_response = {
+            "jsonrpc": "2.0",
+            "id": "text-format-id",
+            "result": [{"output": "Arista cEOSLab\n\nSoftware image version: 4.33.2F-40713977.4332F"}, {"output": "Hostname: DC1-LEAF1A\nFQDN:     DC1-LEAF1A\n"}],
+        }
+
+        response = EapiResponse.from_jsonrpc(jsonrpc_response, request)
+
+        # Verify results contain the text output
+        assert len(response) == 2
+        assert response.results[0].output is not None
+        assert "Arista cEOSLab" in response.results[0].output
+        assert response.results[1].output is not None
+        assert "Hostname: DC1-LEAF1A" in response.results[1].output
+
+    def test_from_jsonrpc_with_timestamps(self) -> None:
+        """Test from_jsonrpc with timestamps enabled."""
+        # Mock request with timestamps
+        request = EapiRequest(commands=["show version"], format=EapiCommandFormat.JSON, timestamps=True)
+
+        # Mock response data with timestamps
+        jsonrpc_response = {
+            "jsonrpc": "2.0",
+            "id": "timestamp-id",
+            "result": [
+                {
+                    "modelName": "cEOSLab",
+                    "version": "4.33.2F-40713977.4332F (engineering build)",
+                    "_meta": {"execStartTime": 1741014072.2534037, "execDuration": 0.0024061203002929688},
+                }
+            ],
+        }
+
+        response = EapiResponse.from_jsonrpc(jsonrpc_response, request)
+
+        # Verify timestamp data is processed
+        assert len(response) == 1
+        assert response.results[0].start_time == 1741014072.2534037
+        assert response.results[0].duration == 0.0024061203002929688
+
+        # Verify _meta is removed from output
+        assert response.results[0].output is not None
+        assert "_meta" not in response.results[0].output
+
+    def test_from_jsonrpc_error_stop_on_error_true(self) -> None:
+        """Test from_jsonrpc with error and stop_on_error=True."""
+        # Mock request with stop_on_error=True
+        request = EapiRequest(commands=["show bad command", "show version", "show hostname"], stop_on_error=True)
+
+        # Mock error response
+        jsonrpc_response = {
+            "jsonrpc": "2.0",
+            "id": "error-id",
+            "error": {
+                "code": 1002,
+                "message": "CLI command 1 of 3 'show bad command' failed: invalid command",
+                "data": [{"errors": ["Invalid input (at token 1: 'bad')"]}],
+            },
+        }
+
+        response = EapiResponse.from_jsonrpc(jsonrpc_response, request)
+
+        # Verify error info
+        assert response.request_id == "error-id"
+        assert response.error_code == 1002
+        assert response.error_message == "CLI command 1 of 3 'show bad command' failed: invalid command"
+        assert response.success is False
+
+        # Verify results - should have entries for all commands
+        assert len(response) == 3
+
+        # First command failed
+        assert response.results[0].command == "show bad command"
+        assert response.results[0].output is None
+        assert response.results[0].success is False
+        assert response.results[0].errors == ["Invalid input (at token 1: 'bad')"]
+
+        # Remaining commands weren't executed due to stop_on_error=True
+        assert response.results[1].command == "show version"
+        assert response.results[1].output is None
+        assert response.results[1].success is False
+        assert "Command not executed due to previous error" in response.results[1].errors
+        assert response.results[1].executed is False
+
+        assert response.results[2].command == "show hostname"
+        assert response.results[2].output is None
+        assert response.results[2].success is False
+        assert "Command not executed due to previous error" in response.results[2].errors
+        assert response.results[2].executed is False
+
+    def test_from_jsonrpc_error_stop_on_error_false(self) -> None:
+        """Test from_jsonrpc with error and stop_on_error=False."""
+        # Mock request with stop_on_error=False
+        request = EapiRequest(commands=["show bad command", "show version", "show hostname"], stop_on_error=False)
+
+        # Mock response with error for first command but others succeed
+        jsonrpc_response = {
+            "jsonrpc": "2.0",
+            "id": "error-continue-id",
+            "error": {
+                "code": 1002,
+                "message": "CLI command 1 of 3 'show bad command' failed: invalid command",
+                "data": [
+                    {"errors": ["Invalid input (at token 1: 'bad')"]},
+                    {"modelName": "cEOSLab", "version": "4.33.2F-40713977.4332F (engineering build)"},
+                    {"hostname": "DC1-LEAF1A"},
+                ],
+            },
+        }
+
+        response = EapiResponse.from_jsonrpc(jsonrpc_response, request)
+
+        # Verify error info
+        assert response.request_id == "error-continue-id"
+        assert response.error_code == 1002
+        assert response.error_message == "CLI command 1 of 3 'show bad command' failed: invalid command"
+        assert response.success is False
+
+        # Verify individual command results
+        assert len(response) == 3
+
+        # First command failed
+        assert response.results[0].command == "show bad command"
+        assert response.results[0].output is None
+        assert response.results[0].success is False
+        assert response.results[0].errors == ["Invalid input (at token 1: 'bad')"]
+
+        # Remaining commands succeeded
+        assert response.results[1].command == "show version"
+        assert response.results[1].output == {"modelName": "cEOSLab", "version": "4.33.2F-40713977.4332F (engineering build)"}
+        assert response.results[1].success is True
+
+        assert response.results[2].command == "show hostname"
+        assert response.results[2].output == {"hostname": "DC1-LEAF1A"}
+        assert response.results[2].success is True
+
+    def test_from_jsonrpc_raise_on_error(self) -> None:
+        """Test from_jsonrpc with raise_on_error=True."""
+        # Mock request
+        request = EapiRequest(commands=["show bad command"])
+
+        # Mock error response
+        jsonrpc_response = {
+            "jsonrpc": "2.0",
+            "id": "raise-error-id",
+            "error": {
+                "code": 1002,
+                "message": "CLI command 1 of 1 'show bad command' failed: invalid command",
+                "data": [{"errors": ["Invalid input (at token 1: 'bad')"]}],
+            },
+        }
+
+        # Should raise EapiReponseError
+        with pytest.raises(EapiReponseError) as excinfo:
+            EapiResponse.from_jsonrpc(jsonrpc_response, request, raise_on_error=True)
+
+        # Verify the exception contains the response
+        assert excinfo.value.response.request_id == "raise-error-id"
+        assert excinfo.value.response.error_code == 1002
+        assert excinfo.value.response.error_message == "CLI command 1 of 1 'show bad command' failed: invalid command"
+
+    def test_from_jsonrpc_string_data(self, caplog: pytest.LogCaptureFixture) -> None:
+        """Test from_jsonrpc with string data response."""
+        caplog.set_level(logging.WARNING)
+
+        # Mock request
+        request = EapiRequest(commands=["show bgp ipv4 unicast summary", "show bad command"])
+
+        # Mock response with JSON string
+        jsonrpc_response = {
+            "jsonrpc": "2.0",
+            "id": "EapiExplorer-1",
+            "error": {
+                "code": 1002,
+                "message": "CLI command 2 of 2 'show bad command' failed: invalid command",
+                "data": [
+                    '{"vrfs":{"default":{"vrf":"default","routerId":"10.1.0.11","asn":"65101","peers":{}}}}\n',
+                    {"errors": ["Invalid input (at token 1: 'bad')"]},
+                ],
+            },
+        }
+
+        response = EapiResponse.from_jsonrpc(jsonrpc_response, request)
+
+        # Verify string was parsed as JSON
+        assert response.results[0].output == {"vrfs": {"default": {"vrf": "default", "routerId": "10.1.0.11", "asn": "65101", "peers": {}}}}
+
+        # Now test with a non-JSON string
+        jsonrpc_response = {
+            "jsonrpc": "2.0",
+            "id": "EapiExplorer-1",
+            "error": {
+                "code": 1002,
+                "message": "CLI command 2 of 2 'show bad command' failed: invalid command",
+                "data": ["This is not JSON", {"errors": ["Invalid input (at token 1: 'bad')"]}],
+            },
+        }
+
+        response = EapiResponse.from_jsonrpc(jsonrpc_response, request)
+
+        # Verify WARNING log message
+        assert "Invalid JSON response for command: show bgp ipv4 unicast summary. Storing as text: This is not JSON" in caplog.text
+
+        # Verify string is kept as is
+        assert response.results[0].output == "This is not JSON"
+
+    def test_from_jsonrpc_complex_commands(self) -> None:
+        """Test from_jsonrpc with complex command structures."""
+        # Mock request with complex commands
+        request = EapiRequest(commands=[{"cmd": "enable", "input": "password"}, "show version"])
+
+        # Mock response
+        jsonrpc_response = {
+            "jsonrpc": "2.0",
+            "id": "complex-cmd-id",
+            "result": [{}, {"modelName": "cEOSLab", "version": "4.33.2F-40713977.4332F (engineering build)"}],
+        }
+
+        response = EapiResponse.from_jsonrpc(jsonrpc_response, request)
+
+        # Verify command strings are extracted correctly
+        assert response.results[0].command == "enable"
+        assert response.results[1].command == "show version"
diff --git a/tests/units/asynceapi/test_data.py b/tests/units/asynceapi/test_data.py
index a1400fe8..94004c1a 100644
--- a/tests/units/asynceapi/test_data.py
+++ b/tests/units/asynceapi/test_data.py
@@ -3,6 +3,7 @@
 # that can be found in the LICENSE file.
 """Unit tests data for the asynceapi client package."""
 
+from asynceapi._constants import EapiCommandFormat
 from asynceapi._types import EapiJsonOutput, JsonRpc
 
 SUCCESS_EAPI_RESPONSE: EapiJsonOutput = {
@@ -86,5 +87,10 @@ ERROR_EAPI_RESPONSE: EapiJsonOutput = {
 }
 """Error eAPI JSON response."""
 
-JSONRPC_REQUEST_TEMPLATE: JsonRpc = {"jsonrpc": "2.0", "method": "runCmds", "params": {"version": 1, "cmds": [], "format": "json"}, "id": "EapiExplorer-1"}
+JSONRPC_REQUEST_TEMPLATE: JsonRpc = {
+    "jsonrpc": "2.0",
+    "method": "runCmds",
+    "params": {"version": 1, "cmds": [], "format": EapiCommandFormat.JSON},
+    "id": "EapiExplorer-1",
+}
 """Template for JSON-RPC eAPI request. `cmds` must be filled by the parametrize decorator."""
diff --git a/tests/units/asynceapi/test_device.py b/tests/units/asynceapi/test_device.py
index b2293d91..c15eb355 100644
--- a/tests/units/asynceapi/test_device.py
+++ b/tests/units/asynceapi/test_device.py
@@ -1,7 +1,7 @@
 # Copyright (c) 2023-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.
-"""Unit tests the asynceapi.device module."""
+"""Unit tests for the asynceapi.device module."""
 
 from __future__ import annotations
 
@@ -63,13 +63,12 @@ async def test_jsonrpc_exec_eapi_command_error(
     jsonrpc_request = JSONRPC_REQUEST_TEMPLATE.copy()
     jsonrpc_request["params"]["cmds"] = cmds
 
-    error_eapi_response = ERROR_EAPI_RESPONSE.copy()
-    httpx_mock.add_response(json=error_eapi_response)
+    httpx_mock.add_response(json=ERROR_EAPI_RESPONSE)
 
     with pytest.raises(EapiCommandError) as exc_info:
         await asynceapi_device.jsonrpc_exec(jsonrpc=jsonrpc_request)
 
-    assert exc_info.value.passed == [error_eapi_response["error"]["data"][0]]
+    assert exc_info.value.passed == [ERROR_EAPI_RESPONSE["error"]["data"][0]]
     assert exc_info.value.failed == "bad command"
     assert exc_info.value.errors == ["Invalid input (at token 1: 'bad')"]
     assert exc_info.value.errmsg == "CLI command 2 of 3 'bad command' failed: invalid command"