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.

Parameters:

Name	Type	Description	Default
device	AntaDevice	Device object to be added.	required

Source code in anta/inventory/__init__.py

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

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

Parameters:

Name	Type	Description	Default
established_only	bool	Whether or not to include only established devices.	False
tags	set[str] | None	Tags to filter devices.	None
devices	set[str] | None	Names to filter devices.	None

Returns:

Type	Description
AntaInventory	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.

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

    Returns
    -------
    AntaInventory
        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.

Parameters:

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

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.

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