diff options
Diffstat (limited to 'asynceapi/device.py')
| -rw-r--r-- | asynceapi/device.py | 291 |
1 files changed, 291 insertions, 0 deletions
diff --git a/asynceapi/device.py b/asynceapi/device.py new file mode 100644 index 0000000..04ec3ab --- /dev/null +++ b/asynceapi/device.py @@ -0,0 +1,291 @@ +# Copyright (c) 2024 Arista Networks, Inc. +# Use of this source code is governed by the Apache License 2.0 +# that can be found in the LICENSE file. +# Initially written by Jeremy Schulman at https://github.com/jeremyschulman/aio-eapi +"""asynceapi.Device definition.""" +# ----------------------------------------------------------------------------- +# System Imports +# ----------------------------------------------------------------------------- + +from __future__ import annotations + +from socket import getservbyname +from typing import TYPE_CHECKING, Any + +# ----------------------------------------------------------------------------- +# Public Imports +# ----------------------------------------------------------------------------- +import httpx + +# ----------------------------------------------------------------------------- +# Private Imports +# ----------------------------------------------------------------------------- +from .aio_portcheck import port_check_url +from .config_session import SessionConfig +from .errors import EapiCommandError + +if TYPE_CHECKING: + from collections.abc import Sequence + +# ----------------------------------------------------------------------------- +# Exports +# ----------------------------------------------------------------------------- + + +__all__ = ["Device"] + + +# ----------------------------------------------------------------------------- +# +# CODE BEGINS +# +# ----------------------------------------------------------------------------- + + +class Device(httpx.AsyncClient): + """ + Represent the async JSON-RPC client that communicates with an Arista EOS device. + + This class inherits directly from the + httpx.AsyncClient, so any initialization options can be passed directly. + """ + + auth = None + EAPI_OFMT_OPTIONS = ("json", "text") + EAPI_DEFAULT_OFMT = "json" + + def __init__( # noqa: PLR0913 # pylint: disable=too-many-arguments + self, + host: str | None = None, + username: str | None = None, + password: str | None = None, + proto: str = "https", + port: str | int | None = None, + **kwargs: Any, # noqa: ANN401 + ) -> None: + """ + Initialize the Device class. + + As a subclass to httpx.AsyncClient, the caller can provide any of those initializers. + Specific parameters for Device class are all optional and described below. + + Parameters + ---------- + host: The EOS target device, either hostname (DNS) or ipaddress. + username: The login user-name; requires the password parameter. + password: The login password; requires the username parameter. + proto: The protocol, http or https, to communicate eAPI with the device. + port: If not provided, the proto value is used to look up the associated + port (http=80, https=443). If provided, overrides the port used to + communite with the device. + + Other Parameters + ---------------- + base_url: str + If provided, the complete URL to the device eAPI endpoint. + + auth: + If provided, used as the httpx authorization initializer value. If + not provided, then username+password is assumed by the Caller and + used to create a BasicAuth instance. + """ + self.port = port or getservbyname(proto) + self.host = host + kwargs.setdefault("base_url", httpx.URL(f"{proto}://{self.host}:{self.port}")) + kwargs.setdefault("verify", False) + + if username and password: + self.auth = httpx.BasicAuth(username, password) + + kwargs.setdefault("auth", self.auth) + + super().__init__(**kwargs) + self.headers["Content-Type"] = "application/json-rpc" + + async def check_connection(self) -> bool: + """ + Check the target device to ensure that the eAPI port is open and accepting connections. + + It is recommended that a Caller checks the connection before involving cli commands, + but this step is not required. + + Returns + ------- + True when the device eAPI is accessible, False otherwise. + """ + return await port_check_url(self.base_url) + + async def cli( # noqa: PLR0913 # pylint: disable=too-many-arguments + self, + command: str | dict[str, Any] | None = None, + commands: Sequence[str | dict[str, Any]] | None = None, + ofmt: str | None = None, + version: int | str | None = "latest", + *, + suppress_error: bool = False, + auto_complete: bool = False, + expand_aliases: bool = False, + req_id: int | str | None = None, + ) -> list[dict[str, Any] | str] | dict[str, Any] | str | None: + """ + Execute one or more CLI commands. + + Parameters + ---------- + command: + A single command to execute; results in a single output response + commands: + A list of commands to execute; results in a list of output responses + ofmt: + Either 'json' or 'text'; indicates the output format for the CLI commands. + version: + By default the eAPI will use "version 1" for all API object models. + This driver will, by default, always set version to "latest" so + that the behavior matches the CLI of the device. The caller can + override the "latest" behavior by explicitly setting the version. + suppress_error: + When not False, then if the execution of the command would-have + raised an EapiCommandError, rather than raising this exception this + routine will return the value None. + + For example, if the following command had raised + EapiCommandError, now response would be set to None instead. + + response = dev.cli(..., suppress_error=True) + auto_complete: + Enabled/disables the command auto-compelete feature of the EAPI. Per the + documentation: + Allows users to use shorthand commands in eAPI calls. With this + parameter included a user can send 'sh ver' via eAPI to get the + output of 'show version'. + expand_aliases: + Enables/disables the command use of User defined alias. Per the + documentation: + Allowed users to provide the expandAliases parameter to eAPI + calls. This allows users to use aliased commands via the API. + 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. + req_id: + A unique identifier that will be echoed back by the switch. May be a string or number. + + Returns + ------- + One or List of output responses, per the description above. + """ + if not any((command, commands)): + msg = "Required 'command' or 'commands'" + raise RuntimeError(msg) + + jsonrpc = self._jsonrpc_command( + commands=[command] if command else commands, ofmt=ofmt, version=version, auto_complete=auto_complete, expand_aliases=expand_aliases, req_id=req_id + ) + + try: + res = await self.jsonrpc_exec(jsonrpc) + return res[0] if command else res + except EapiCommandError: + if suppress_error: + return None + raise + + def _jsonrpc_command( # noqa: PLR0913 # pylint: disable=too-many-arguments + self, + commands: Sequence[str | dict[str, Any]] | None = None, + ofmt: str | None = None, + version: int | str | None = "latest", + *, + auto_complete: bool = False, + expand_aliases: bool = False, + req_id: int | str | None = None, + ) -> dict[str, Any]: + """Create the JSON-RPC command dictionary object.""" + cmd: dict[str, Any] = { + "jsonrpc": "2.0", + "method": "runCmds", + "params": { + "version": version, + "cmds": commands, + "format": ofmt or self.EAPI_DEFAULT_OFMT, + }, + "id": req_id or id(self), + } + if auto_complete is not None: + cmd["params"].update({"autoComplete": auto_complete}) + + if expand_aliases is not None: + cmd["params"].update({"expandAliases": expand_aliases}) + + return cmd + + async def jsonrpc_exec(self, jsonrpc: dict[str, Any]) -> list[dict[str, Any] | str]: + """ + Execute the JSON-RPC dictionary object. + + Parameters + ---------- + jsonrpc: + The JSON-RPC as created by the `meth`:_jsonrpc_command(). + + Raises + ------ + EapiCommandError + In the event that a command resulted in an error response. + + Returns + ------- + The list of command results; either dict or text depending on the + JSON-RPC format parameter. + """ + res = await self.post("/command-api", json=jsonrpc) + res.raise_for_status() + body = res.json() + + commands = jsonrpc["params"]["cmds"] + ofmt = jsonrpc["params"]["format"] + + get_output = (lambda _r: _r["output"]) if ofmt == "text" else (lambda _r: _r) + + # if there are no errors then return the list of command results. + if (err_data := body.get("error")) is None: + return [get_output(cmd_res) for cmd_res in body["result"]] + + # --------------------------------------------------------------------- + # if we are here, then there were some command errors. Raise a + # EapiCommandError exception with args (commands that failed, passed, + # not-executed). + # --------------------------------------------------------------------- + + # -------------------------- eAPI specification ---------------------- + # On an error, no result object is present, only an error object, which + # is guaranteed to have the following attributes: code, messages, and + # data. Similar to the result object in the successful response, the + # data object is a list of objects corresponding to the results of all + # commands up to, and including, the failed command. If there was a an + # error before any commands were executed (e.g. bad credentials), data + # will be empty. The last object in the data array will always + # correspond to the failed command. The command failure details are + # always stored in the errors array. + + cmd_data = err_data["data"] + len_data = len(cmd_data) + err_at = len_data - 1 + err_msg = err_data["message"] + + raise EapiCommandError( + passed=[get_output(cmd_data[cmd_i]) for cmd_i, cmd in enumerate(commands[:err_at])], + failed=commands[err_at]["cmd"], + errors=cmd_data[err_at]["errors"], + errmsg=err_msg, + not_exec=commands[err_at + 1 :], + ) + + def config_session(self, name: str) -> SessionConfig: + """ + return a SessionConfig instance bound to this device with the given session name. + + Parameters + ---------- + name: The config-session name + """ + return SessionConfig(self, name) |