feat(anta): NetBox as an inventory source

anta/cli/get/__init__.py

@@ -15,6 +15,7 @@ def get() -> None:
 
 get.add_command(commands.from_cvp)
 get.add_command(commands.from_ansible)
+get.add_command(commands.from_netbox)
 get.add_command(commands.inventory)
 get.add_command(commands.tags)
 get.add_command(commands.tests)

anta/cli/get/commands.py

@@ -22,7 +22,7 @@
 from anta.cli.get.utils import inventory_output_options
 from anta.cli.utils import ExitCode, inventory_options
 
-from .utils import create_inventory_from_ansible, create_inventory_from_cvp, explore_package, get_cv_token
+from .utils import create_inventory_from_ansible, create_inventory_from_cvp, create_inventory_from_netbox, explore_package, get_cv_token
 
 if TYPE_CHECKING:
     from anta.inventory import AntaInventory
@@ -106,6 +106,24 @@ def from_ansible(ctx: click.Context, output: Path, ansible_group: str, ansible_i
         ctx.exit(ExitCode.USAGE_ERROR)
 
 
+@click.command
+@click.pass_context
+@inventory_output_options
+@click.option("--nb-instance", "-nbi", help="Name of the NetBox instance", type=str, required=True)
+@click.option("--nb-token", "-nbt", help="NetBox token", type=str, required=True)
+@click.option("--nb-platform", "-nbp", help="NetBox device platform", type=str, default="Arista EOS", required=False)
+@click.option("--nb-site", "-nbs", help="NetBox site (case sensitive)", type=str, default=None, required=False)
+@click.option("--nb-verify", "-nbf", help="NetBox verify SSL", type=bool, default=False, required=False)
+def from_netbox(ctx: click.Context, nb_instance: str, output: Path, nb_token: str, nb_platform: str, nb_site: str | None = None, nb_verify: bool = False) -> None:
+    """Build ANTA inventory from a NetBox instance."""
+    logger.info("Building inventory from netbox instance file '%s'", nb_instance)
+    try:
+        create_inventory_from_netbox(nb_instance=nb_instance, output=output, token=nb_token, platform=nb_platform, site=nb_site, verify=nb_verify)
+    except ValueError as e:
+        logger.error(str(e))
+        ctx.exit(ExitCode.USAGE_ERROR)
+
+
 @click.command
 @inventory_options
 @click.option("--connected/--not-connected", help="Display inventory after connection has been created", default=False, required=False)

anta/cli/get/utils.py

@@ -8,6 +8,7 @@
 import functools
 import importlib
 import inspect
+import ipaddress
 import json
 import logging
 import pkgutil
@@ -214,6 +215,56 @@ def create_inventory_from_ansible(inventory: Path, output: Path, ansible_group:
     write_inventory_to_file(ansible_hosts, output)
 
 
+def create_inventory_from_netbox(nb_instance: str, output: Path, token: str, platform: str = "Arista EOS", site: str | None = None, verify: bool = False) -> None:
+    """Fetch devices from NetBox filtered by a specific platform.
+
+    Parameters
+    ----------
+    nb_instance
+        The NetBox API instance.
+    output
+        ANTA inventory file to generate.
+    token
+        The token used to authenticate to the NetBox instance.
+    platform
+        The platform to filter devices by.
+    site
+        The site to filter devices by.
+    verify
+        Verify the SSL certification of the NetBox instance.
+
+    """
+    session = requests.session()
+    session.verify = verify
+    try:
+        import pynetbox
+    except ImportError as e:
+        logging.error(e)
+
+    try:
+        # Initialize NetBox API
+        nb = pynetbox.api(nb_instance, token=token)
+
+        # Platform name to filter
+        platform = nb.dcim.platforms.get(q=[platform])
+
+        devices = nb.dcim.devices.filter(platform=platform.slug, site=site) if site else nb.dcim.devices.filter(platform=platform.slug)
+
+        inventory = []
+        for device in devices:
+            host_entry = {
+                "host": str(ipaddress.ip_interface(device.primary_ip).ip),
+                "name": device.name,
+                "tags": [tag.name for tag in device.tags],
+            }
+            inventory.append(host_entry)
+
+        write_inventory_to_file(inventory, output)
+
+    except Exception as e:
+        raise ValueError(e) from e
+
+
 def explore_package(module_name: str, test_name: str | None = None, *, short: bool = False, count: bool = False) -> int:
     """Parse ANTA test submodules recursively and print AntaTest examples.
 

pyproject.toml

@@ -100,6 +100,9 @@ doc = [
   "black>=24.10.0",
   "mkdocs-github-admonitions-plugin>=0.0.3"
 ]
+netbox = [
+  "pynetbox>=7.4.1"
+]
 
 [project.urls]
 Homepage = "https://anta.arista.com"