Inventory module

AntaInventory

Bases: dict[str, AntaDevice]

Inventory abstraction for ANTA framework.

devices property

devices: list[AntaDevice]

List of AntaDevice in this inventory.

__setitem__

__setitem__(key: str, value: AntaDevice) -> None

Set a device in the inventory.

Source code in anta/inventory/__init__.py

def __setitem__(self, key: str, value: AntaDevice) -> None:
    """Set a device in the inventory."""
    if key != value.name:
        msg = f"The key must be the device name for device '{value.name}'. Use AntaInventory.add_device()."
        raise RuntimeError(msg)
    return super().__setitem__(key, value)

add_device

add_device(device: AntaDevice) -> None

Add a device to final inventory.

Args:

device: Device object to be added

Source code in anta/inventory/__init__.py

def add_device(self, device: AntaDevice) -> None:
    """Add a device to final inventory.

    Args:
    ----
        device: Device object to be added

    """
    self[device.name] = device

connect_inventory async

connect_inventory() -> None

Run refresh() coroutines for all AntaDevice objects in this inventory.

Source code in anta/inventory/__init__.py

async def connect_inventory(self) -> None:
    """Run `refresh()` coroutines for all AntaDevice objects in this inventory."""
    logger.debug("Refreshing devices...")
    results = await asyncio.gather(
        *(device.refresh() for device in self.values()),
        return_exceptions=True,
    )
    for r in results:
        if isinstance(r, Exception):
            message = "Error when refreshing inventory"
            anta_log_exception(r, message, logger)

get_inventory

get_inventory(*, established_only: bool = False, tags: set[str] | None = None, devices: set[str] | None = None) -> AntaInventory

Return a filtered inventory.

Args:

established_only: Whether or not to include only established devices.
tags: Tags to filter devices.
devices: Names to filter devices.

Returns:

Type	Description
An inventory with filtered AntaDevice objects.

Source code in anta/inventory/__init__.py

def get_inventory(self, *, established_only: bool = False, tags: set[str] | None = None, devices: set[str] | None = None) -> AntaInventory:
    """Return a filtered inventory.

    Args:
    ----
        established_only: Whether or not to include only established devices.
        tags: Tags to filter devices.
        devices: Names to filter devices.

    Returns
    -------
        An inventory with filtered AntaDevice objects.
    """

    def _filter_devices(device: AntaDevice) -> bool:
        """Select the devices based on the inputs `tags`, `devices` and `established_only`."""
        if tags is not None and all(tag not in tags for tag in device.tags):
            return False
        if devices is None or device.name in devices:
            return bool(not established_only or device.established)
        return False

    filtered_devices: list[AntaDevice] = list(filter(_filter_devices, self.values()))
    result = AntaInventory()
    for device in filtered_devices:
        result.add_device(device)
    return result

parse staticmethod

parse(filename: str | Path, username: str, password: str, enable_password: str | None = None, timeout: float | None = None, *, enable: bool = False, insecure: bool = False, disable_cache: bool = False) -> AntaInventory

Create an AntaInventory instance from an inventory file.

The inventory devices are AsyncEOSDevice instances.

Args:

filename: Path to device inventory YAML file.
username: Username to use to connect to devices.
password: Password to use to connect to devices.
enable_password: Enable password to use if required.
timeout: Timeout value in seconds for outgoing API calls.
enable: Whether or not the commands need to be run in enable mode towards the devices.
insecure: Disable SSH Host Key validation.
disable_cache: Disable cache globally.

Raises:

Type	Description
InventoryRootKeyError: Root key of inventory is missing.	InventoryIncorrectSchemaError: Inventory file is not following AntaInventory Schema.

Source code in anta/inventory/__init__.py

@staticmethod
def parse(
    filename: str | Path,
    username: str,
    password: str,
    enable_password: str | None = None,
    timeout: float | None = None,
    *,
    enable: bool = False,
    insecure: bool = False,
    disable_cache: bool = False,
) -> AntaInventory:
    """Create an AntaInventory instance from an inventory file.

    The inventory devices are AsyncEOSDevice instances.

    Args:
    ----
        filename: Path to device inventory YAML file.
        username: Username to use to connect to devices.
        password: Password to use to connect to devices.
        enable_password: Enable password to use if required.
        timeout: Timeout value in seconds for outgoing API calls.
        enable: Whether or not the commands need to be run in enable mode towards the devices.
        insecure: Disable SSH Host Key validation.
        disable_cache: Disable cache globally.

    Raises
    ------
        InventoryRootKeyError: Root key of inventory is missing.
        InventoryIncorrectSchemaError: Inventory file is not following AntaInventory Schema.

    """
    inventory = AntaInventory()
    kwargs: dict[str, Any] = {
        "username": username,
        "password": password,
        "enable": enable,
        "enable_password": enable_password,
        "timeout": timeout,
        "insecure": insecure,
        "disable_cache": disable_cache,
    }
    if username is None:
        message = "'username' is required to create an AntaInventory"
        logger.error(message)
        raise ValueError(message)
    if password is None:
        message = "'password' is required to create an AntaInventory"
        logger.error(message)
        raise ValueError(message)

    try:
        filename = Path(filename)
        with filename.open(encoding="UTF-8") as file:
            data = safe_load(file)
    except (TypeError, YAMLError, OSError) as e:
        message = f"Unable to parse ANTA Device Inventory file '{filename}'"
        anta_log_exception(e, message, logger)
        raise

    if AntaInventory.INVENTORY_ROOT_KEY not in data:
        exc = InventoryRootKeyError(f"Inventory root key ({AntaInventory.INVENTORY_ROOT_KEY}) is not defined in your inventory")
        anta_log_exception(exc, f"Device inventory is invalid! (from {filename})", logger)
        raise exc

    try:
        inventory_input = AntaInventoryInput(**data[AntaInventory.INVENTORY_ROOT_KEY])
    except ValidationError as e:
        anta_log_exception(e, f"Device inventory is invalid! (from {filename})", logger)
        raise

    # Read data from input
    AntaInventory._parse_hosts(inventory_input, inventory, **kwargs)
    AntaInventory._parse_networks(inventory_input, inventory, **kwargs)
    AntaInventory._parse_ranges(inventory_input, inventory, **kwargs)

    return inventory

exceptions

Manage Exception in Inventory module.

InventoryIncorrectSchemaError

Bases: Exception

Error when user data does not follow ANTA schema.

InventoryRootKeyError

Bases: Exception

Error raised when inventory root key is not found.