Device

AntaDevice base class

UML representation

AntaDevice

AntaDevice(name: str, tags: Optional[list[str]] = 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	str	Device name
is_online	bool	True if the device IP is reachable and a port can be open
established	bool	True if remote command execution succeeds
hw_model	Optional[str]	Hardware model of the device
tags	list[str]	List of tags for this device
cache	Optional[Cache]	In-memory cache from aiocache library for this device (None if cache is disabled)
cache_locks	Optional[defaultdict[str, Lock]]	Dictionary mapping keys to asyncio locks to guarantee exclusive access to the cache if not disabled

Parameters:

Name	Type	Description	Default
name	str	Device name	required
tags	Optional[list[str]]	List of tags for this device	None
disable_cache	bool	Disable caching for all commands for this device. Defaults to False.	False

Source code in anta/device.py

def __init__(self, name: str, tags: Optional[list[str]] = None, disable_cache: bool = False) -> None:
    """
    Constructor of AntaDevice

    Args:
        name: Device name
        tags: List of tags for this device
        disable_cache: Disable caching for all commands for this device. Defaults to False.
    """
    self.name: str = name
    self.hw_model: Optional[str] = None
    self.tags: list[str] = tags if tags is not None else []
    # A device always has its own name as tag
    self.tags.append(self.name)
    self.is_online: bool = False
    self.established: bool = False
    self.cache: Optional[Cache] = None
    self.cache_locks: Optional[defaultdict[str, asyncio.Lock]] = 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) -> None

Collects 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.

Parameters:

Name	Type	Description	Default
command	AntaCommand	The command to process.	required

Source code in anta/device.py

async def collect(self, command: AntaCommand) -> None:
    """
    Collects 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 (AntaCommand): The command to process.
    """
    # 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(f"Cache hit for {command.command} on {self.name}")
                command.output = cached_output
            else:
                await self._collect(command=command)
                await self.cache.set(command.uid, command.output)  # pylint: disable=no-member
    else:
        await self._collect(command=command)

collect_commands async

collect_commands(commands: list[AntaCommand]) -> None

Collect multiple commands.

Parameters:

Name	Type	Description	Default
commands	list[AntaCommand]	the commands to collect	required

Source code in anta/device.py

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

    Args:
        commands: the commands to collect
    """
    await asyncio.gather(*(self.collect(command=command) 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.

Parameters:

Name	Type	Description	Default
sources	list[Path]	List of files to copy to or from the device.	required
destination	Path	Local or remote destination when copying the files. Can be a folder.	required
direction	Literal['to', 'from']	Defines if this coroutine copies files to or from the device.	'from'

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.
    """
    raise NotImplementedError(f"copy() method has not been implemented in {self.__class__.__name__} definition")

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
    """

supports

supports(command: AntaCommand) -> bool

Returns True if the command is supported on the device hardware platform, False otherwise.

Source code in anta/device.py

def supports(self, command: AntaCommand) -> bool:
    """Returns True if the command is supported on the device hardware platform, False otherwise."""
    unsupported = any("not supported on this hardware platform" in e for e in command.errors)
    logger.debug(command)
    if unsupported:
        logger.debug(f"{command.command} is not supported on {self.hw_model}")
    return not unsupported

Async EOS device class

UML representation

AsyncEOSDevice

AsyncEOSDevice(host: str, username: str, password: str, name: Optional[str] = None, enable: bool = False, enable_password: Optional[str] = None, port: Optional[int] = None, ssh_port: Optional[int] = 22, tags: Optional[list[str]] = None, timeout: Optional[float] = None, insecure: bool = False, proto: Literal['http', 'https'] = 'https', 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		List of tags for this device

Parameters:

Name	Type	Description	Default
host	str	Device FQDN or IP	required
username	str	Username to connect to eAPI and SSH	required
password	str	Password to connect to eAPI and SSH	required
name	Optional[str]	Device name	None
enable	bool	Device needs privileged access	False
enable_password	Optional[str]	Password used to gain privileged access on EOS	None
port	Optional[int]	eAPI port. Defaults to 80 is proto is ‘http’ or 443 if proto is ‘https’.	None
ssh_port	Optional[int]	SSH port	22
tags	Optional[list[str]]	List of tags for this device	None
timeout	Optional[float]	Timeout value in seconds for outgoing connections. Default to 10 secs.	None
insecure	bool	Disable SSH Host Key validation	False
proto	Literal['http', 'https']	eAPI protocol. Value can be ‘http’ or ‘https’	'https'
disable_cache	bool	Disable caching for all commands for this device. Defaults to False.	False

Source code in anta/device.py

def __init__(  # pylint: disable=R0913
    self,
    host: str,
    username: str,
    password: str,
    name: Optional[str] = None,
    enable: bool = False,
    enable_password: Optional[str] = None,
    port: Optional[int] = None,
    ssh_port: Optional[int] = 22,
    tags: Optional[list[str]] = None,
    timeout: Optional[float] = None,
    insecure: bool = False,
    proto: Literal["http", "https"] = "https",
    disable_cache: bool = False,
) -> None:
    """
    Constructor of 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: Device needs privileged access
        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: List of tags for this device
        timeout: Timeout value in seconds for outgoing connections. Default to 10 secs.
        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. Defaults to False.
    """
    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)
    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: aioeapi.Device = aioeapi.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, **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().

Parameters:

Name	Type	Description	Default
sources	list[Path]	List of files to copy to or from the device.	required
destination	Path	Local or remote destination when copying the files. Can be a folder.	required
direction	Literal['to', 'from']	Defines if this coroutine copies files to or from the device.	'from'

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: Union[list[tuple[SSHClientConnection, Path]], list[Path]]
        dst: Union[tuple[SSHClientConnection, Path], Path]
        if direction == "from":
            src = [(conn, file) for file in sources]
            dst = destination
            for file in sources:
                logger.info(f"Copying '{file}' from device {self.name} to '{destination}' locally")

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

        else:
            logger.critical(f"'direction' argument to copy() fonction is invalid: {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(f"Refreshing device {self.name}")
    self.is_online = await self._session.check_connection()
    if self.is_online:
        COMMAND: str = "show version"
        HW_MODEL_KEY: str = "modelName"
        try:
            response = await self._session.cli(command=COMMAND)
        except aioeapi.EapiCommandError as e:
            logger.warning(f"Cannot get hardware information from device {self.name}: {e.errmsg}")

        except (HTTPError, ConnectError) as e:
            logger.warning(f"Cannot get hardware information from device {self.name}: {exc_to_str(e)}")

        else:
            if HW_MODEL_KEY in response:
                self.hw_model = response[HW_MODEL_KEY]
            else:
                logger.warning(f"Cannot get hardware information from device {self.name}: cannot parse '{COMMAND}'")

    else:
        logger.warning(f"Could not connect to device {self.name}: cannot open eAPI port")

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