-
Notifications
You must be signed in to change notification settings - Fork 39
Description
Please disregard if the team feels this change is not required.
At the moment we require a file to load an inventory. This is true in the CLI and using ANTA directly in Python. There are ways around this in the Python implementation but I feel there should be a more simple way.
The current way to implement this in Python may look something like below:
my_inv = {
"anta_inventory": {
"hosts": [
{"host": "192.168.0.10", "name": "s1-spine1"},
{"host": "192.168.0.11", "name": "s1-spine2"},
]
}
}
async def main(inv: AntaInventory) -> None:
"""Read an AntaInventory and try to connect to every device in the inventory.
Print a message for every device connection status
"""
await inv.connect_inventory()
for device in inv.values():
if device.established:
print(f"Device {device.name} is online")
else:
print(f"Could not connect to device {device.name}")
if __name__ == "__main__":
USERNAME = "arista"
PASSWORD = "arista"
inventory = AntaInventory()
# Get the list of hosts from your dictionary
host_list = my_inv["anta_inventory"]["hosts"]
print(f"Loading {len(host_list)} devices from the dictionary...")
# Iterate over the list of hosts
for host_data in host_list:
# Create an AsyncEOSDevice instance for each host
device = AsyncEOSDevice(
host=host_data["host"],
name=host_data["name"],
username=USERNAME,
password=PASSWORD
)
# Add the device to the inventory
inventory.add_device(device)
# # Run the main coroutine
res = asyncio.run(main(inventory))This works great but requires the user to do more work than required. Loading from a generic Python Dictionary would allow integration with any third-party inventory management system. The only ask on the user is to massage the data into the current ANTA requirement or skip down one level to a lists of hosts with the required dictionary data per host. Like the my_inv variable shown above. The user could then skip having to create a temporary file to load the inventory or manually create the structure using AsyncEOSDevice.
I tested one variation of this implementation by adjusting the parse function
@staticmethod
def parse(
inventory_input: str | Path | dict[str, Any],
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:
data: dict[str, Any]
input_name: str # For logging purposes
if isinstance(inventory_input, dict):
data = inventory_input
input_name = "dictionary input"
elif isinstance(inventory_input, (str, Path)):
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)
filename = Path(inventory_input)
input_name = str(filename)
try:
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
else:
raise TypeError(f"inventory_input must be a filename (str or Path) or a dictionary, not {type(inventory_input)}")
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 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 {input_name})", logger)
raise exc
try:
# Renamed variable here to avoid confusion with the method's inventory_input parameter
inventory_input_model = AntaInventoryInput(**data[AntaInventory.INVENTORY_ROOT_KEY])
except ValidationError as e:
anta_log_exception(e, f"Device inventory is invalid! (from {input_name})", logger)
raise
# Read data from input
AntaInventory._parse_hosts(inventory_input_model, inventory, **kwargs)
AntaInventory._parse_networks(inventory_input_model, inventory, **kwargs)
AntaInventory._parse_ranges(inventory_input_model, inventory, **kwargs)
return inventory