Inventory module

AntaInventory

Bases: dict

Inventory abstraction for ANTA framework.

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.

    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: Optional[list[str]] = None) -> AntaInventory

Returns a filtered inventory.

Parameters:

Name	Type	Description	Default
established_only	bool	Whether or not to include only established devices. Default False.	False
tags	Optional[list[str]]	List of tags to filter devices.	None

Returns:

Name	Type	Description
AntaInventory	AntaInventory	An inventory with filtered AntaDevice objects.

Source code in anta/inventory/__init__.py

def get_inventory(self, established_only: bool = False, tags: Optional[list[str]] = None) -> AntaInventory:
    """
    Returns a filtered inventory.

    Args:
        established_only: Whether or not to include only established devices. Default False.
        tags: List of tags to filter devices.

    Returns:
        AntaInventory: An inventory with filtered AntaDevice objects.
    """

    def _filter_devices(device: AntaDevice) -> bool:
        """
        Helper function to select the devices based on the input tags
        and the requirement for an established connection.
        """
        if tags is not None and all(tag not in tags for tag in device.tags):
            return False
        return bool(not established_only or device.established)

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

parse staticmethod

parse(filename: str | Path, username: str, password: str, enable: bool = False, enable_password: Optional[str] = None, timeout: Optional[float] = None, 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 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	bool	Whether or not the commands need to be run in enable mode towards the devices	False
enable_password	str	Enable password to use if required	None
timeout	float	timeout in seconds for every API call.	None
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.
InventoryIncorrectSchema	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: bool = False,
    enable_password: Optional[str] = None,
    timeout: Optional[float] = None,
    insecure: bool = False,
    disable_cache: bool = False,
) -> AntaInventory:
    # pylint: disable=too-many-arguments
    """
    Create an AntaInventory instance from an inventory file.
    The inventory devices are AsyncEOSDevice instances.

    Args:
        filename (str): Path to device inventory YAML file
        username (str): Username to use to connect to devices
        password (str): Password to use to connect to devices
        enable (bool): Whether or not the commands need to be run in enable mode towards the devices
        enable_password (str, optional): Enable password to use if required
        timeout (float, optional): timeout in seconds for every API call.
        insecure (bool): Disable SSH Host Key validation
        disable_cache (bool): Disable cache globally

    Raises:
        InventoryRootKeyError: Root key of inventory is missing.
        InventoryIncorrectSchema: 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:
        with open(file=filename, mode="r", 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.

InventoryIncorrectSchema

Bases: Exception

Error when user data does not follow ANTA schema.

InventoryRootKeyError

Bases: Exception

Error raised when inventory root key is not found.