as-python-lib.md

ANTA is a Python library that can be used in user applications. This section describes how you can leverage ANTA Python modules to help you create your own NRFU solution.

!!! tip If you are unfamiliar with asyncio, refer to the Python documentation relevant to your Python version - https://docs.python.org/3/library/asyncio.html

AntaDevice Abstract Class

A device is represented in ANTA as a instance of a subclass of the AntaDevice abstract class. There are few abstract methods that needs to be implemented by child classes:

• The collect() coroutine is in charge of collecting outputs of AntaCommand instances.
• The refresh() coroutine is in charge of updating attributes of the AntaDevice instance. These attributes are used by AntaInventory to filter out unreachable devices or by AntaTest to skip devices based on their hardware models.

The copy() coroutine is used to copy files to and from the device. It does not need to be implemented if tests are not using it.

AsyncEOSDevice Class

The AsyncEOSDevice class is an implementation of AntaDevice for Arista EOS. It uses the aio-eapi eAPI client and the AsyncSSH library.

• The _collect() coroutine collects AntaCommand outputs using eAPI.
• The refresh() coroutine tries to open a TCP connection on the eAPI port and update the is_online attribute accordingly. If the TCP connection succeeds, it sends a show version command to gather the hardware model of the device and updates the established and hw_model attributes.
• The copy() coroutine copies files to and from the device using the SCP protocol.

AntaInventory Class

The AntaInventory class is a subclass of the standard Python type dict. The keys of this dictionary are the device names, the values are AntaDevice instances.

AntaInventory provides methods to interact with the ANTA inventory:

• The add_device() method adds an AntaDevice instance to the inventory. Adding an entry to AntaInventory with a key different from the device name is not allowed.
• The get_inventory() returns a new AntaInventory instance with filtered out devices based on the method inputs.
• The connect_inventory() coroutine will execute the refresh() coroutines of all the devices in the inventory.
• The parse() static method creates an AntaInventory instance from a YAML file and returns it. The devices are AsyncEOSDevice instances.

Examples

Parse an ANTA inventory file

This script parses an ANTA inventory file, connects to devices and print their status

"""
Example
"""
import asyncio

from anta.inventory import AntaInventory


async def main(inv: AntaInventory) -> None:
    """
    Take an AntaInventory and:
    1. try to connect to every device in the inventory
    2. print a message for every device connection status
    """
    await inv.connect_inventory()

    for device in inv.values():
        if device.established:
            print(f"Device {device.name} is online")
        else:
            print(f"Could not connect to device {device.name}")

if __name__ == "__main__":
    # Create the AntaInventory instance
    inventory = AntaInventory.parse(
        filename="inv.yml",
        username="arista",
        password="@rista123",
    )

    # Run the main coroutine
    res = asyncio.run(main(inventory))

??? note "How to create your inventory file" Please visit this dedicated section for how to use inventory and catalog files.

Run EOS commands

This script runs a list of EOS commands on reachable devices

"""
Example
"""
# This is needed to run the script for python < 3.10 for typing annotations
from __future__ import annotations

import asyncio
from pprint import pprint

from anta.inventory import AntaInventory
from anta.models import AntaCommand


async def main(inv: AntaInventory, commands: list[str]) -> dict[str, list[AntaCommand]]:
    """
    Take an AntaInventory and a list of commands as string and:
    1. try to connect to every device in the inventory
    2. collect the results of the commands from each device

    Returns:
      a dictionary where key is the device name and the value is the list of AntaCommand ran towards the device
    """
    await inv.connect_inventory()

    # Make a list of coroutine to run commands towards each connected device
    coros = []
    # dict to keep track of the commands per device
    result_dict = {}
    for name, device in inv.get_inventory(established_only=True).items():
        anta_commands = [AntaCommand(command=command, ofmt="json") for command in commands]
        result_dict[name] = anta_commands
        coros.append(device.collect_commands(anta_commands))

    # Run the coroutines
    await asyncio.gather(*coros)

    return result_dict


if __name__ == "__main__":
    # Create the AntaInventory instance
    inventory = AntaInventory.parse(
        filename="inv.yml",
        username="arista",
        password="@rista123",
    )

    # Create a list of commands with json output
    commands = ["show version", "show ip bgp summary"]

    # Run the main asyncio  entry point
    res = asyncio.run(main(inventory, commands))

    pprint(res)