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(inventory_file: str, username: str, password: str, enable: bool = False, enable_password: Optional[str] = None, timeout: Optional[float] = None, insecure: bool = False) -> AntaInventory

Create an AntaInventory instance from an inventory file. The inventory devices are AsyncEOSDevice instances.

Parameters:

Name	Type	Description	Default
inventory_file	str	Path to inventory YAML file where user has described his inputs	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
timeout	float	timeout in seconds for every API call.	None

Raises:

Type	Description
InventoryRootKeyError	Root key of inventory is missing.
InventoryIncorrectSchema	Inventory file is not following AntaInventory Schema.
InventoryUnknownFormat	Output format is not supported.

Source code in anta/inventory/__init__.py

@staticmethod
def parse(
    inventory_file: str,
    username: str,
    password: str,
    enable: bool = False,
    enable_password: Optional[str] = None,
    timeout: Optional[float] = None,
    insecure: bool = False,
) -> AntaInventory:
    # pylint: disable=too-many-arguments
    """
    Create an AntaInventory instance from an inventory file.
    The inventory devices are AsyncEOSDevice instances.

    Args:
        inventory_file (str): Path to inventory YAML file where user has described his inputs
        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
        timeout (float, optional): timeout in seconds for every API call.

    Raises:
        InventoryRootKeyError: Root key of inventory is missing.
        InventoryIncorrectSchema: Inventory file is not following AntaInventory Schema.
        InventoryUnknownFormat: Output format is not supported.
    """

    inventory = AntaInventory()
    kwargs: dict[str, Any] = {
        "username": username,
        "password": password,
        "enable": enable,
        "enable_password": enable_password,
        "timeout": timeout,
        "insecure": insecure,
    }
    kwargs = {k: v for k, v in kwargs.items() if v is not None}

    with open(inventory_file, "r", encoding="UTF-8") as file:
        data = safe_load(file)

    # Load data using Pydantic
    try:
        inventory_input = AntaInventoryInput(**data[AntaInventory.INVENTORY_ROOT_KEY])
    except KeyError as exc:
        logger.error(f"Inventory root key is missing: {AntaInventory.INVENTORY_ROOT_KEY}")
        raise InventoryRootKeyError(f"Inventory root key ({AntaInventory.INVENTORY_ROOT_KEY}) is not defined in your inventory") from exc
    except ValidationError as exc:
        logger.error("Inventory data are not compliant with inventory models")
        raise InventoryIncorrectSchema(f"Inventory is not following the schema: {str(exc)}") from exc

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

---

Last update: August 18, 2023