AntaDevice base class

UML representation

AntaDevice

AntaDevice(name: str, tags: set[str] | None = None, *, disable_cache: bool = False)

Bases: ABC

Abstract class representing a device in ANTA.

An implementation of this class must override the abstract coroutines _collect() and refresh().

Attributes:

Name	Type	Description
name	Device name	is_online: True if the device IP is reachable and a port can be open. established: True if remote command execution succeeds. hw_model: Hardware model of the device. tags: Tags for this device. cache: In-memory cache from aiocache library for this device (None if cache is disabled). cache_locks: Dictionary mapping keys to asyncio locks to guarantee exclusive access to the cache if not disabled.

Args:

name: Device name.
tags: Tags for this device.
disable_cache: Disable caching for all commands for this device.

Source code in anta/device.py

def __init__(self, name: str, tags: set[str] | None = None, *, disable_cache: bool = False) -> None:
    """Initialize an AntaDevice.

    Args:
    ----
        name: Device name.
        tags: Tags for this device.
        disable_cache: Disable caching for all commands for this device.

    """
    self.name: str = name
    self.hw_model: str | None = None
    self.tags: set[str] = tags if tags is not None else set()
    # A device always has its own name as tag
    self.tags.add(self.name)
    self.is_online: bool = False
    self.established: bool = False
    self.cache: Cache | None = None
    self.cache_locks: defaultdict[str, asyncio.Lock] | None = None

    # Initialize cache if not disabled
    if not disable_cache:
        self._init_cache()

cache_statistics property

cache_statistics: dict[str, Any] | None

Returns the device cache statistics for logging purposes.

__hash__

__hash__() -> int

Implement hashing for AntaDevice objects.

Source code in anta/device.py

def __hash__(self) -> int:
    """Implement hashing for AntaDevice objects."""
    return hash(self._keys)

collect async

collect(command: AntaCommand, *, collection_id: str | None = None) -> None

Collect the output for a specified command.

When caching is activated on both the device and the command, this method prioritizes retrieving the output from the cache. In cases where the output isn’t cached yet, it will be freshly collected and then stored in the cache for future access. The method employs asynchronous locks based on the command’s UID to guarantee exclusive access to the cache.

When caching is NOT enabled, either at the device or command level, the method directly collects the output via the private _collect method without interacting with the cache.

Args:

command: The command to collect.
collection_id: An identifier used to build the eAPI request ID.

Source code in anta/device.py

async def collect(self, command: AntaCommand, *, collection_id: str | None = None) -> None:
    """Collect the output for a specified command.

    When caching is activated on both the device and the command,
    this method prioritizes retrieving the output from the cache. In cases where the output isn't cached yet,
    it will be freshly collected and then stored in the cache for future access.
    The method employs asynchronous locks based on the command's UID to guarantee exclusive access to the cache.

    When caching is NOT enabled, either at the device or command level, the method directly collects the output
    via the private `_collect` method without interacting with the cache.

    Args:
    ----
        command: The command to collect.
        collection_id: An identifier used to build the eAPI request ID.
    """
    # Need to ignore pylint no-member as Cache is a proxy class and pylint is not smart enough
    # https://github.com/pylint-dev/pylint/issues/7258
    if self.cache is not None and self.cache_locks is not None and command.use_cache:
        async with self.cache_locks[command.uid]:
            cached_output = await self.cache.get(command.uid)  # pylint: disable=no-member

            if cached_output is not None:
                logger.debug("Cache hit for %s on %s", command.command, self.name)
                command.output = cached_output
            else:
                await self._collect(command=command, collection_id=collection_id)
                await self.cache.set(command.uid, command.output)  # pylint: disable=no-member
    else:
        await self._collect(command=command, collection_id=collection_id)

collect_commands async

collect_commands(commands: list[AntaCommand], *, collection_id: str | None = None) -> None

Collect multiple commands.

Args:

commands: The commands to collect.
collection_id: An identifier used to build the eAPI request ID.

Source code in anta/device.py

async def collect_commands(self, commands: list[AntaCommand], *, collection_id: str | None = None) -> None:
    """Collect multiple commands.

    Args:
    ----
        commands: The commands to collect.
        collection_id: An identifier used to build the eAPI request ID.
    """
    await asyncio.gather(*(self.collect(command=command, collection_id=collection_id) for command in commands))

copy async

copy(sources: list[Path], destination: Path, direction: Literal['to', 'from'] = 'from') -> None

Copy files to and from the device, usually through SCP.

It is not mandatory to implement this for a valid AntaDevice subclass.

Args:

sources: List of files to copy to or from the device.
destination: Local or remote destination when copying the files. Can be a folder.
direction: Defines if this coroutine copies files to or from the device.

Source code in anta/device.py

async def copy(self, sources: list[Path], destination: Path, direction: Literal["to", "from"] = "from") -> None:
    """Copy files to and from the device, usually through SCP.

    It is not mandatory to implement this for a valid AntaDevice subclass.

    Args:
    ----
        sources: List of files to copy to or from the device.
        destination: Local or remote destination when copying the files. Can be a folder.
        direction: Defines if this coroutine copies files to or from the device.

    """
    _ = (sources, destination, direction)
    msg = f"copy() method has not been implemented in {self.__class__.__name__} definition"
    raise NotImplementedError(msg)

refresh abstractmethod async

refresh() -> None

Update attributes of an AntaDevice instance.

This coroutine must update the following attributes of AntaDevice: - is_online: When the device IP is reachable and a port can be open - established: When a command execution succeeds - hw_model: The hardware model of the device

Source code in anta/device.py

@abstractmethod
async def refresh(self) -> None:
    """Update attributes of an AntaDevice instance.

    This coroutine must update the following attributes of AntaDevice:
        - `is_online`: When the device IP is reachable and a port can be open
        - `established`: When a command execution succeeds
        - `hw_model`: The hardware model of the device
    """

Async EOS device class

UML representation

AsyncEOSDevice

AsyncEOSDevice(host: str, username: str, password: str, name: str | None = None, enable_password: str | None = None, port: int | None = None, ssh_port: int | None = 22, tags: set[str] | None = None, timeout: float | None = None, proto: Literal['http', 'https'] = 'https', *, enable: bool = False, insecure: bool = False, disable_cache: bool = False)

Bases: AntaDevice

Implementation of AntaDevice for EOS using aio-eapi.

Attributes:

Name	Type	Description
name	Device name	is_online: True if the device IP is reachable and a port can be open established: True if remote command execution succeeds hw_model: Hardware model of the device tags: Tags for this device

Args:

host: Device FQDN or IP.
username: Username to connect to eAPI and SSH.
password: Password to connect to eAPI and SSH.
name: Device name.
enable: Collect commands using privileged mode.
enable_password: Password used to gain privileged access on EOS.
port: eAPI port. Defaults to 80 is proto is 'http' or 443 if proto is 'https'.
ssh_port: SSH port.
tags: Tags for this device.
timeout: Timeout value in seconds for outgoing API calls.
insecure: Disable SSH Host Key validation.
proto: eAPI protocol. Value can be 'http' or 'https'.
disable_cache: Disable caching for all commands for this device.

Source code in anta/device.py

def __init__(
    self,
    host: str,
    username: str,
    password: str,
    name: str | None = None,
    enable_password: str | None = None,
    port: int | None = None,
    ssh_port: int | None = 22,
    tags: set[str] | None = None,
    timeout: float | None = None,
    proto: Literal["http", "https"] = "https",
    *,
    enable: bool = False,
    insecure: bool = False,
    disable_cache: bool = False,
) -> None:
    """Instantiate an AsyncEOSDevice.

    Args:
    ----
        host: Device FQDN or IP.
        username: Username to connect to eAPI and SSH.
        password: Password to connect to eAPI and SSH.
        name: Device name.
        enable: Collect commands using privileged mode.
        enable_password: Password used to gain privileged access on EOS.
        port: eAPI port. Defaults to 80 is proto is 'http' or 443 if proto is 'https'.
        ssh_port: SSH port.
        tags: Tags for this device.
        timeout: Timeout value in seconds for outgoing API calls.
        insecure: Disable SSH Host Key validation.
        proto: eAPI protocol. Value can be 'http' or 'https'.
        disable_cache: Disable caching for all commands for this device.

    """
    if host is None:
        message = "'host' is required to create an AsyncEOSDevice"
        logger.error(message)
        raise ValueError(message)
    if name is None:
        name = f"{host}{f':{port}' if port else ''}"
    super().__init__(name, tags, disable_cache=disable_cache)
    if username is None:
        message = f"'username' is required to instantiate device '{self.name}'"
        logger.error(message)
        raise ValueError(message)
    if password is None:
        message = f"'password' is required to instantiate device '{self.name}'"
        logger.error(message)
        raise ValueError(message)
    self.enable = enable
    self._enable_password = enable_password
    self._session: asynceapi.Device = asynceapi.Device(host=host, port=port, username=username, password=password, proto=proto, timeout=timeout)
    ssh_params: dict[str, Any] = {}
    if insecure:
        ssh_params["known_hosts"] = None
    self._ssh_opts: SSHClientConnectionOptions = SSHClientConnectionOptions(
        host=host, port=ssh_port, username=username, password=password, client_keys=CLIENT_KEYS, **ssh_params
    )

copy async

copy(sources: list[Path], destination: Path, direction: Literal['to', 'from'] = 'from') -> None

Copy files to and from the device using asyncssh.scp().

Args:

sources: List of files to copy to or from the device.
destination: Local or remote destination when copying the files. Can be a folder.
direction: Defines if this coroutine copies files to or from the device.

Source code in anta/device.py

async def copy(self, sources: list[Path], destination: Path, direction: Literal["to", "from"] = "from") -> None:
    """Copy files to and from the device using asyncssh.scp().

    Args:
    ----
        sources: List of files to copy to or from the device.
        destination: Local or remote destination when copying the files. Can be a folder.
        direction: Defines if this coroutine copies files to or from the device.

    """
    async with asyncssh.connect(
        host=self._ssh_opts.host,
        port=self._ssh_opts.port,
        tunnel=self._ssh_opts.tunnel,
        family=self._ssh_opts.family,
        local_addr=self._ssh_opts.local_addr,
        options=self._ssh_opts,
    ) as conn:
        src: list[tuple[SSHClientConnection, Path]] | list[Path]
        dst: tuple[SSHClientConnection, Path] | Path
        if direction == "from":
            src = [(conn, file) for file in sources]
            dst = destination
            for file in sources:
                message = f"Copying '{file}' from device {self.name} to '{destination}' locally"
                logger.info(message)

        elif direction == "to":
            src = sources
            dst = conn, destination
            for file in src:
                message = f"Copying '{file}' to device {self.name} to '{destination}' remotely"
                logger.info(message)

        else:
            logger.critical("'direction' argument to copy() function is invalid: %s", direction)

            return
        await asyncssh.scp(src, dst)

refresh async

refresh() -> None

Update attributes of an AsyncEOSDevice instance.

This coroutine must update the following attributes of AsyncEOSDevice: - is_online: When a device IP is reachable and a port can be open - established: When a command execution succeeds - hw_model: The hardware model of the device

Source code in anta/device.py

async def refresh(self) -> None:
    """Update attributes of an AsyncEOSDevice instance.

    This coroutine must update the following attributes of AsyncEOSDevice:
    - is_online: When a device IP is reachable and a port can be open
    - established: When a command execution succeeds
    - hw_model: The hardware model of the device
    """
    logger.debug("Refreshing device %s", self.name)
    self.is_online = await self._session.check_connection()
    if self.is_online:
        show_version = AntaCommand(command="show version")
        await self._collect(show_version)
        if not show_version.collected:
            logger.warning("Cannot get hardware information from device %s", self.name)
        else:
            self.hw_model = show_version.json_output.get("modelName", None)
            if self.hw_model is None:
                logger.critical("Cannot parse 'show version' returned by device %s", self.name)
    else:
        logger.warning("Could not connect to device %s: cannot open eAPI port", self.name)

    self.established = bool(self.is_online and self.hw_model)