feat: tenant, named, vrf, domain, cluster and site filters

one filter option plus its value, set in config or via
NETBOX_INVENTORY_FILTER / NETBOX_INVENTORY_FILTER_VALUE:
- tenant (both sources), resolved by name server-side
- named (ip-addresses), only addresses with a DNS name, deduplicated
- vrf (ip-addresses), resolved by name server-side
- domain (both sources), hostname ends with the domain
- cluster, site (virtual-machines), resolved by name server-side

netbox_inventory.py

@@ -25,6 +25,16 @@ SOURCE_IP_ADDRESSES = "ip-addresses"
 SOURCE_VIRTUAL_MACHINES = "virtual-machines"
 SOURCES = (SOURCE_IP_ADDRESSES, SOURCE_VIRTUAL_MACHINES)
 
+# filter option -> sources it applies to
+FILTERS = {
+    "tenant": SOURCES,
+    "domain": SOURCES,
+    "named": (SOURCE_IP_ADDRESSES,),
+    "vrf": (SOURCE_IP_ADDRESSES,),
+    "cluster": (SOURCE_VIRTUAL_MACHINES,),
+    "site": (SOURCE_VIRTUAL_MACHINES,),
+}
+
 
 class InventoryError(Exception):
     """Raised when the inventory cannot be built (configuration or API problems)."""
@@ -111,6 +121,18 @@ class NetBoxClient:
             params = None  # the "next" URL already carries the query string
         return results
 
+    def resolve_id(self, path: str, name: str, label: str) -> int:
+        """
+        Look up an object by name and return its numeric id.
+
+        Raises:
+            NetBoxAPIError: If no object with that name exists.
+        """
+        matches = self.get_all(path, params={"name": name})
+        if not matches:
+            raise NetBoxAPIError(f'No {label} named "{name}" found in NetBox.')
+        return matches[0]["id"]
+
 
 def load_config(path: str | None = None) -> dict[str, Any]:
     """
@@ -162,6 +184,95 @@ def load_config(path: str | None = None) -> dict[str, Any]:
     return merged
 
 
+def parse_bool(value: str) -> bool:
+    """
+    Parse a yes/no filter value, an empty string counts as yes.
+
+    Raises:
+        InventoryError: If the value is not recognisable as a boolean.
+    """
+    lowered = value.strip().lower()
+    if lowered in ("", "1", "true", "yes", "on"):
+        return True
+    if lowered in ("0", "false", "no", "off"):
+        return False
+    raise InventoryError(f'Filter value "{value}" is not a valid boolean.')
+
+
+def domain_matches(hostname: str, domain: str) -> bool:
+    """
+    Check whether a hostname belongs to a domain.
+
+    The hostname must end with the domain on a label boundary, so
+    "web.example.com" matches "example.com" but "notexample.com" does not.
+    """
+    if not hostname:
+        return False
+    domain = domain.lstrip(".").rstrip(".").lower()
+    hostname = hostname.rstrip(".").lower()
+    return hostname == domain or hostname.endswith("." + domain)
+
+
+def validate_options(source: str, filter_name: str, filter_value: str) -> None:
+    """
+    Validate the source/filter combination before touching the API.
+
+    Raises:
+        InventoryError: On unknown sources or filters, a filter that does not
+            apply to the source, or a missing filter value.
+    """
+    if source not in SOURCES:
+        raise InventoryError(f'Unknown source "{source}", valid sources: {", ".join(SOURCES)}.')
+    if not filter_name:
+        return
+    if filter_name not in FILTERS:
+        raise InventoryError(
+            f'Unknown filter "{filter_name}", valid filters: {", ".join(FILTERS)}.'
+        )
+    if source not in FILTERS[filter_name]:
+        raise InventoryError(f'Filter "{filter_name}" does not apply to source "{source}".')
+    if filter_name != "named" and not filter_value:
+        raise InventoryError(f'Filter "{filter_name}" requires a value.')
+
+
+def server_side_params(
+    client: NetBoxClient, filter_name: str, filter_value: str
+) -> dict[str, Any]:
+    """Translate name-based filters into NetBox API query parameters."""
+    if filter_name == "tenant":
+        return {"tenant_id": client.resolve_id("tenancy/tenants", filter_value, "tenant")}
+    if filter_name == "vrf":
+        return {"vrf_id": client.resolve_id("ipam/vrfs", filter_value, "VRF")}
+    if filter_name == "cluster":
+        return {"cluster_id": client.resolve_id("virtualization/clusters", filter_value, "cluster")}
+    if filter_name == "site":
+        return {"site_id": client.resolve_id("dcim/sites", filter_value, "site")}
+    return {}
+
+
+def apply_client_side_filters(
+    entries: list[tuple[str, dict[str, Any]]], filter_name: str, filter_value: str
+) -> list[tuple[str, dict[str, Any]]]:
+    """
+    Apply the filters that cannot be expressed as API query parameters.
+
+    "domain" keeps hostnames belonging to the given domain, "named" keeps only
+    IP addresses with a DNS name and deduplicates them by hostname.
+    """
+    if filter_name == "domain":
+        return [(name, host) for name, host in entries if domain_matches(name, filter_value)]
+    if filter_name == "named" and parse_bool(filter_value):
+        seen: set[str] = set()
+        named = []
+        for name, host in entries:
+            if not host.get("netbox_dns_name") or name in seen:
+                continue
+            seen.add(name)
+            named.append((name, host))
+        return named
+    return entries
+
+
 def strip_prefix(address: str) -> str:
     """Return a bare IP address without the CIDR prefix length."""
     return address.split("/", 1)[0]
@@ -207,12 +318,15 @@ def vm_host_entry(vm: dict[str, Any]) -> tuple[str, dict[str, Any]]:
     return name, hostvars
 
 
-def fetch_entries(client: NetBoxClient, source: str) -> list[tuple[str, dict[str, Any]]]:
+def fetch_entries(
+    client: NetBoxClient, source: str, params: dict[str, Any] | None = None
+) -> list[tuple[str, dict[str, Any]]]:
     """Fetch all hosts of the configured source from NetBox."""
     if source == SOURCE_IP_ADDRESSES:
-        records = client.get_all("ipam/ip-addresses")
+        records = client.get_all("ipam/ip-addresses", params)
         return [ip_host_entry(record) for record in records if record.get("address")]
-    records = client.get_all("virtualization/virtual-machines", {"exclude": "config_context"})
+    params = {"exclude": "config_context", **(params or {})}
+    records = client.get_all("virtualization/virtual-machines", params)
     return [vm_host_entry(record) for record in records if record.get("name")]
 
 
@@ -221,14 +335,15 @@ def build_inventory(config: dict[str, Any], client: NetBoxClient) -> dict[str, A
     Build the full --list inventory structure for Ansible.
 
     Raises:
-        InventoryError: On an unknown source or NetBox API failures.
+        InventoryError: On invalid source/filter options or NetBox API failures.
     """
     source = config["source"]
-    if source not in SOURCES:
+    filter_name = config["filter"]
-        raise InventoryError(
+    filter_value = config["filter_value"]
-            f'Unknown source "{source}", valid sources: {", ".join(SOURCES)}.'
+    validate_options(source, filter_name, filter_value)
-        )
+    params = server_side_params(client, filter_name, filter_value) if filter_name else {}
-    entries = fetch_entries(client, source)
+    entries = fetch_entries(client, source, params)
+    entries = apply_client_side_filters(entries, filter_name, filter_value)
     hostvars: dict[str, dict[str, Any]] = {}
     for name, host in entries:
         if name in hostvars: