Skip to content

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
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
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
88
89
90
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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
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
170
171
172
173
174
175
176
177
178
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
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
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
180
181
182
183
184
185
186
187
188
@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
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
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
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
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
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
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)