ANTA Inventory API

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

    Parameters
    ----------
    device
        Device object to be added.

    """
    self[device.name] = device

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)

async def disconnect_inventory(self) -> None:
    """Run `disconnect()` coroutines for all AntaDevice objects in this inventory."""
    results = await asyncio.gather(
        *(device.disconnect() for device in self.values()),
        return_exceptions=True,
    )
    for r in results:
        if isinstance(r, Exception):
            logger.warning("Error when disconnecting inventory: %s", exc_to_str(r))

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

    Each hosts is dumped individually.
    """
    hosts = [
        AntaInventoryHost(
            name=device.name,
            host=device.host if not self.is_base_class(device) else device.name,
            port=device.port if not self.is_base_class(device) else None,
            tags=device.tags,
            disable_cache=device.cache is None,
            use_session_auth=device.use_session_auth if isinstance(device, AsyncEOSDevice) else False,
            **({"ssl_params": device.ssl_params} if isinstance(device, AsyncEOSDevice) and device.ssl_params is not None else {}),
        )
        for device in self.devices
    ]
    return AntaInventoryInput(hosts=hosts)

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

def is_base_class(self, device: AntaDevice) -> TypeIs[AntaDevice]:
    """Check the type of device, return True if the device is an AntaDevice."""
    return not hasattr(device, "host") and not hasattr(device, "port")

@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,
    use_session_auth: bool | None = None,
) -> 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
        Global timeout value in seconds for outgoing eAPI calls. None means no timeout.
    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.
    use_session_auth
        Session authentication override. ``True`` forces session auth on for all devices,
        ``False`` (``--no-session-auth``) forces it off regardless of inventory settings,
        ``None`` (unset) defers to the per-device inventory value.

    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, use_session_auth_override=use_session_auth, **kwargs)
    AntaInventory._parse_networks(inventory_input, inventory, use_session_auth_override=use_session_auth, **kwargs)
    AntaInventory._parse_ranges(inventory_input, inventory, use_session_auth_override=use_session_auth, **kwargs)

    return inventory

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(exclude_unset=True, indent=2)

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)