ANTA Inventory API

AntaInventory

Bases: dict[str, AntaDevice]

Inventory abstraction for ANTA framework.

devices property

devices: list[AntaDevice]

List of AntaDevice in this inventory.

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)

dump

dump() -> AntaInventoryInput

Dump the AntaInventory to an AntaInventoryInput.

Each hosts is dumped individually.

Source code in anta/inventory/__init__.py

def dump(self) -> AntaInventoryInput:
    """Dump the AntaInventory to an AntaInventoryInput.

    Each hosts is dumped individually.
    """
    hosts = [
        AntaInventoryHost(
            name=device.name,
            host=device.host if hasattr(device, "host") else device.name,
            port=device.port if hasattr(device, "port") else None,
            tags=device.tags,
            disable_cache=device.cache is None,
        )
        for device in self.devices
    ]
    return AntaInventoryInput(hosts=hosts)

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,
    file_format: Literal["yaml", "json"] = "yaml",
    *,
    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
file_format	Literal['yaml', 'json']	Whether the inventory file is in JSON or YAML.	'yaml'
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,
    file_format: Literal["yaml", "json"] = "yaml",
    *,
    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.
    file_format
        Whether the inventory file is in JSON or YAML.
    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.

    """
    if file_format not in ["yaml", "json"]:
        message = f"'{file_format}' is not a valid format for an AntaInventory file. Only 'yaml' and 'json' are supported."
        raise ValueError(message)

    inventory = AntaInventory()
    kwargs: dict[str, Any] = {
        "username": username,
        "password": password,
        "enable": enable,
        "enable_password": enable_password,
        "timeout": timeout,
        "insecure": insecure,
        "disable_cache": disable_cache,
    }

    try:
        filename = Path(filename)
        with filename.open(encoding="UTF-8") as file:
            data = safe_load(file) if file_format == "yaml" else json_load(file)
    except (TypeError, YAMLError, OSError, ValueError) 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

AntaInventoryInput

Bases: BaseModel

Device inventory input model.

to_json

to_json() -> str

Return a JSON representation string of this model.

Returns:

Type	Description
The JSON representation string of this model.

Source code in anta/inventory/models.py

def to_json(self) -> str:
    """Return a JSON representation string of this model.

    Returns
    -------
        The JSON representation string of this model.
    """
    return self.model_dump_json(serialize_as_any=True, exclude_unset=True, indent=2)

yaml

yaml() -> str

Return a YAML representation string of this model.

Returns:

Type	Description
str	The YAML representation string of this model.

Source code in anta/inventory/models.py

def yaml(self) -> str:
    """Return a YAML representation string of this model.

    Returns
    -------
    str
        The YAML representation string of this model.
    """
    # TODO: Pydantic and YAML serialization/deserialization is not supported natively.
    # This could be improved.
    # https://github.com/pydantic/pydantic/issues/1043
    # Explore if this worth using this: https://github.com/NowanIlfideme/pydantic-yaml
    return yaml.safe_dump(yaml.safe_load(self.model_dump_json(serialize_as_any=True, exclude_unset=True)), width=math.inf)

AntaInventoryHost

Bases: AntaInventoryBaseModel

Host entry of AntaInventoryInput.

Attributes:

Name	Type	Description
host	Hostname | IPvAnyAddress	IP Address or FQDN of the device.
port	Port | None	Custom eAPI port to use.
name	str | None	Custom name of the device.
tags	set[str]	Tags of the device.
disable_cache	bool	Disable cache for this device.

AntaInventoryNetwork

Bases: AntaInventoryBaseModel

Network entry of AntaInventoryInput.

Attributes:

Name	Type	Description
network	IPvAnyNetwork	Subnet to use for scanning.
tags	set[str]	Tags of the devices in this network.
disable_cache	bool	Disable cache for all devices in this network.

AntaInventoryRange

Bases: AntaInventoryBaseModel

IP Range entry of AntaInventoryInput.

Attributes:

Name	Type	Description
start	IPvAnyAddress	IPv4 or IPv6 address for the beginning of the range.
stop	IPvAnyAddress	IPv4 or IPv6 address for the end of the range.
tags	set[str]	Tags of the devices in this IP range.
disable_cache	bool	Disable cache for all devices in this IP range.

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.