From 7912dacf9511df0663e072bf63250211d5e56f79 Mon Sep 17 00:00:00 2001
From: Bryan Joshua Pedini <bryan@pedini.dev>
Date: Tue, 25 Aug 2026 13:09:09 +0200
Subject: [PATCH] feat: configuration loading and netbox api client

config.yaml next to the script (or $NETBOX_INVENTORY_CONFIG) with environment
variable overrides, requests session with token auth, paginated list fetching
and readable error reporting.
---
 .gitignore          |  18 +++++
 config.yaml.example |  15 ++++
 netbox_inventory.py | 186 ++++++++++++++++++++++++++++++++++++++++++++
 ruff.toml           |  12 +++
 4 files changed, 231 insertions(+)
 create mode 100644 .gitignore
 create mode 100644 config.yaml.example
 create mode 100755 netbox_inventory.py
 create mode 100644 ruff.toml

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..5ae67fe
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,18 @@
+# Python
+__pycache__/
+*.py[cod]
+.venv/
+
+# Testing
+.pytest_cache/
+.ruff_cache/
+
+# Configuration
+config.yaml
+
+# IDE
+.idea/
+.vscode/
+
+# OS
+.DS_Store
diff --git a/config.yaml.example b/config.yaml.example
new file mode 100644
index 0000000..b8e4cb4
--- /dev/null
+++ b/config.yaml.example
@@ -0,0 +1,15 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+# Copyright (C) 2026 Bryan Joshua Pedini
+
+netbox:
+  url: "https://netbox.example.com"
+  token: "0123456789abcdef0123456789abcdef01234567"
+  verify_ssl: true
+  timeout: 30
+
+inventory:
+  # where to get the hosts from: ip-addresses | virtual-machines
+  source: "ip-addresses"
+  # optional filter, see README.md for the full list
+  filter: ""
+  filter_value: ""
diff --git a/netbox_inventory.py b/netbox_inventory.py
new file mode 100755
index 0000000..95b2cc3
--- /dev/null
+++ b/netbox_inventory.py
@@ -0,0 +1,186 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: GPL-2.0-or-later
+# Copyright (C) 2026 Bryan Joshua Pedini
+"""
+Ansible dynamic inventory script that sources hosts from NetBox.
+
+Hosts are read from either the NetBox IP address list or the virtual machine
+list, optionally narrowed down by a single filter option; everything is set in
+the configuration file and can be overridden through environment variables.
+"""
+
+import argparse
+import json
+import os
+import sys
+from typing import Any
+
+import requests
+import yaml
+
+DEFAULT_CONFIG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.yaml")
+GROUP_NAME = "netbox"
+
+SOURCE_IP_ADDRESSES = "ip-addresses"
+SOURCE_VIRTUAL_MACHINES = "virtual-machines"
+SOURCES = (SOURCE_IP_ADDRESSES, SOURCE_VIRTUAL_MACHINES)
+
+
+class InventoryError(Exception):
+    """Raised when the inventory cannot be built (configuration or API problems)."""
+
+
+class NetBoxAPIError(InventoryError):
+    """
+    Raised when the NetBox API cannot be reached or returns an error.
+
+    Attributes:
+        message: Human readable description of the failure.
+        status_code: HTTP status code of the response, if one was received.
+    """
+
+    def __init__(self, message: str, status_code: int | None = None):
+        super().__init__(message)
+        self.message = message
+        self.status_code = status_code
+
+
+def _extract_error_message(response: requests.Response) -> str:
+    """Pull a readable error message out of a NetBox error response."""
+    try:
+        data = response.json()
+        if isinstance(data, dict) and "detail" in data:
+            return f"NetBox API error: {data['detail']}"
+    except ValueError:
+        pass
+    return f"NetBox API error: HTTP {response.status_code}"
+
+
+class NetBoxClient:
+    """Thin wrapper around the NetBox REST API."""
+
+    def __init__(self, url: str, token: str, verify_ssl: bool = True, timeout: int = 30):
+        self._url = url.rstrip("/")
+        self._timeout = timeout
+        self._session = requests.Session()
+        self._session.verify = verify_ssl
+        self._session.headers.update(
+            {
+                "Authorization": f"Token {token}",
+                "Accept": "application/json",
+            }
+        )
+
+    def _request(self, url: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
+        """
+        Perform a GET request and return the decoded JSON body.
+
+        Raises:
+            NetBoxAPIError: On connection problems, non-200 responses or
+                unparsable bodies.
+        """
+        try:
+            response = self._session.get(url, params=params, timeout=self._timeout)
+        except requests.exceptions.Timeout as exc:
+            raise NetBoxAPIError("NetBox API request timed out.") from exc
+        except requests.exceptions.SSLError as exc:
+            raise NetBoxAPIError("SSL error while connecting to the NetBox API.") from exc
+        except requests.exceptions.ConnectionError as exc:
+            raise NetBoxAPIError("Unable to connect to the NetBox API.") from exc
+        except requests.exceptions.RequestException as exc:
+            raise NetBoxAPIError(f"NetBox API request failed: {exc}") from exc
+        if response.status_code != 200:
+            raise NetBoxAPIError(
+                _extract_error_message(response), status_code=response.status_code
+            )
+        try:
+            return response.json()
+        except ValueError as exc:
+            raise NetBoxAPIError(
+                "NetBox API returned invalid JSON.", status_code=response.status_code
+            ) from exc
+
+    def get_all(self, path: str, params: dict[str, Any] | None = None) -> list[dict[str, Any]]:
+        """Return every result of a paginated list endpoint."""
+        results: list[dict[str, Any]] = []
+        url: str | None = f"{self._url}/api/{path.strip('/')}/"
+        while url:
+            data = self._request(url, params=params)
+            results.extend(data.get("results", []))
+            url = data.get("next")
+            params = None  # the "next" URL already carries the query string
+        return results
+
+
+def load_config(path: str | None = None) -> dict[str, Any]:
+    """
+    Load the YAML configuration file and apply environment overrides.
+
+    Args:
+        path: Configuration file path; falls back to $NETBOX_INVENTORY_CONFIG,
+            then to config.yaml next to this script.
+
+    Returns:
+        A flat configuration dict with url, token, verify_ssl, timeout,
+        source, filter and filter_value keys.
+
+    Raises:
+        InventoryError: If the file is malformed or url/token are missing.
+    """
+    path = path or os.environ.get("NETBOX_INVENTORY_CONFIG") or DEFAULT_CONFIG_FILE
+    config: dict[str, Any] = {}
+    if os.path.exists(path):
+        with open(path, encoding="utf-8") as handle:
+            loaded = yaml.safe_load(handle) or {}
+        if not isinstance(loaded, dict):
+            raise InventoryError(f"Invalid configuration file: {path}")
+        config = loaded
+    netbox = config.get("netbox") or {}
+    inventory = config.get("inventory") or {}
+    merged = {
+        "url": os.environ.get("NETBOX_URL", netbox.get("url") or ""),
+        "token": os.environ.get("NETBOX_TOKEN", netbox.get("token") or ""),
+        "verify_ssl": bool(netbox.get("verify_ssl", True)),
+        "timeout": netbox.get("timeout", 30),
+        "source": os.environ.get(
+            "NETBOX_INVENTORY_SOURCE", inventory.get("source") or SOURCE_IP_ADDRESSES
+        ),
+        "filter": os.environ.get("NETBOX_INVENTORY_FILTER", inventory.get("filter") or ""),
+        "filter_value": os.environ.get(
+            "NETBOX_INVENTORY_FILTER_VALUE", inventory.get("filter_value") or ""
+        ),
+    }
+    try:
+        merged["timeout"] = int(merged["timeout"])
+    except (TypeError, ValueError) as exc:
+        raise InventoryError("Configuration value netbox.timeout must be a number.") from exc
+    if not merged["url"] or not merged["token"]:
+        raise InventoryError(
+            "NetBox url and token are required, set them in the configuration file "
+            "or through the NETBOX_URL / NETBOX_TOKEN environment variables."
+        )
+    return merged
+
+
+def main() -> int:
+    """Entry point for the Ansible dynamic inventory protocol."""
+    parser = argparse.ArgumentParser(description="Ansible dynamic inventory backed by NetBox.")
+    group = parser.add_mutually_exclusive_group(required=True)
+    group.add_argument("--list", action="store_true", help="output the full inventory as JSON")
+    group.add_argument("--host", metavar="HOSTNAME", help="output variables for a single host")
+    args = parser.parse_args()
+    if args.host:
+        # all host variables are returned through _meta in --list
+        print(json.dumps({}))
+        return 0
+    try:
+        load_config()
+        print(json.dumps({GROUP_NAME: {"hosts": []}, "_meta": {"hostvars": {}}}, indent=2))
+    except InventoryError as exc:
+        print(f"netbox-inventory: {exc}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/ruff.toml b/ruff.toml
new file mode 100644
index 0000000..d83b6e0
--- /dev/null
+++ b/ruff.toml
@@ -0,0 +1,12 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+# Copyright (C) 2026 Bryan Joshua Pedini
+
+line-length = 100
+target-version = "py310"
+
+[lint]
+select = ["E", "F", "W", "I", "N", "UP", "B", "C4", "SIM"]
+ignore = ["E501"]
+
+[lint.pydocstyle]
+convention = "google"