netbox_client.py•11.4 kB
#!/usr/bin/env python3
"""
NetBox Client Library
This module provides a base class for NetBox client implementations and a REST API implementation.
"""
import abc
from typing import Any, Dict, List, Optional, Union
import requests
from urllib.parse import urlencode
import logging
# logging.basicConfig(level=logging.INFO)
# Configure logging
# Configure logging to write to a file
logging.basicConfig(
filename='log.txt', # Log file name
level=logging.INFO, # Minimum level to log
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
class NetBoxClientBase(abc.ABC):
"""
Abstract base class for NetBox client implementations.
This class defines the interface for CRUD operations that can be implemented
either via the REST API or directly via the ORM in a NetBox plugin.
"""
@abc.abstractmethod
def get(self, endpoint: str, id: Optional[int] = None, params: Optional[Dict[str, Any]] = None) -> Union[Dict[str, Any], List[Dict[str, Any]]]:
"""
Retrieve one or more objects from NetBox.
Args:
endpoint: The API endpoint (e.g., 'dcim/sites', 'ipam/prefixes')
id: Optional ID to retrieve a specific object
params: Optional query parameters for filtering
Returns:
Either a single object dict or a list of object dicts
"""
pass
@abc.abstractmethod
def create(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]:
"""
Create a new object in NetBox.
Args:
endpoint: The API endpoint (e.g., 'dcim/sites', 'ipam/prefixes')
data: Object data to create
Returns:
The created object as a dict
"""
pass
@abc.abstractmethod
def update(self, endpoint: str, id: int, data: Dict[str, Any]) -> Dict[str, Any]:
"""
Update an existing object in NetBox.
Args:
endpoint: The API endpoint (e.g., 'dcim/sites', 'ipam/prefixes')
id: ID of the object to update
data: Object data to update
Returns:
The updated object as a dict
"""
pass
@abc.abstractmethod
def delete(self, endpoint: str, id: int) -> bool:
"""
Delete an object from NetBox.
Args:
endpoint: The API endpoint (e.g., 'dcim/sites', 'ipam/prefixes')
id: ID of the object to delete
Returns:
True if deletion was successful, False otherwise
"""
pass
@abc.abstractmethod
def bulk_create(self, endpoint: str, data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Create multiple objects in NetBox.
Args:
endpoint: The API endpoint (e.g., 'dcim/sites', 'ipam/prefixes')
data: List of object data to create
Returns:
List of created objects as dicts
"""
pass
@abc.abstractmethod
def bulk_update(self, endpoint: str, data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Update multiple objects in NetBox.
Args:
endpoint: The API endpoint (e.g., 'dcim/sites', 'ipam/prefixes')
data: List of object data to update (must include ID)
Returns:
List of updated objects as dicts
"""
pass
@abc.abstractmethod
def bulk_delete(self, endpoint: str, ids: List[int]) -> bool:
"""
Delete multiple objects from NetBox.
Args:
endpoint: The API endpoint (e.g., 'dcim/sites', 'ipam/prefixes')
ids: List of IDs to delete
Returns:
True if deletion was successful, False otherwise
"""
pass
class NetBoxRestClient(NetBoxClientBase):
"""
NetBox client implementation using the REST API.
"""
def __init__(self, url: str, token: str, verify_ssl: bool = False):
"""
Initialize the REST API client.
Args:
url: The base URL of the NetBox instance
token: API token for authentication
verify_ssl: Whether to verify SSL certificates
"""
self.base_url = url.rstrip('/')
self.api_url = f"{self.base_url}/api"
self.token = token
self.verify_ssl = verify_ssl
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Token {token}',
'Content-Type': 'application/json',
'Accept': 'application/json',
})
def _build_url(self, endpoint: str, id: Optional[int] = None, params: Optional[Dict[str, Any]] = None) -> str:
"""Build the full URL for an API request."""
endpoint = endpoint.strip('/')
if id is not None:
return f"{self.api_url}/{endpoint}/{id}/"
return f"{self.api_url}/{endpoint}/"
def get_count(self, endpoint: str, id: Optional[int] = None, params: Optional[Dict[str, Any]] = None) -> Union[Dict[str, Any], List[Dict[str, Any]]]:
"""
Retrieve the count of objects from NetBox via the REST API.
Args:
endpoint: The API endpoint (e.g., 'dcim/sites', 'ipam/prefixes')
id: Optional ID to retrieve a specific object (not applicable for count)
params: Optional query parameters for filtering
Returns:
A dictionary containing the count of objects
Raises:
requests.HTTPError: If the request fails
"""
url = self._build_url(endpoint, id)
logging.info("GET count object to URL in netbox: %s", url)
logging.info(f"GET count object to params: {params}")
response = self.session.get(url, params=params, verify=self.verify_ssl)
# response = self.session.get(url, verify=self.verify_ssl)
response.raise_for_status()
data = response.json()
return data
def get(self, endpoint: str, id: Optional[int] = None, params: Optional[Dict[str, Any]] = None) -> Union[Dict[str, Any], List[Dict[str, Any]]]:
"""
Retrieve one or more objects from NetBox via the REST API.
Args:
endpoint: The API endpoint (e.g., 'dcim/sites', 'ipam/prefixes')
id: Optional ID to retrieve a specific object
params: Optional query parameters for filtering
Returns:
Either a single object dict or a list of object dicts
Raises:
requests.HTTPError: If the request fails
"""
url = self._build_url(endpoint, id)
logging.info("GET request to URL in netbox: %s", url)
logging.info(f"GET request to params: {params}")
response = self.session.get(url, params=params, verify=self.verify_ssl)
# response = self.session.get(url, verify=self.verify_ssl)
response.raise_for_status()
data = response.json()
if id is None and 'results' in data:
# Handle paginated results
return data['results']
return data
def create(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]:
"""
Create a new object in NetBox via the REST API.
Args:
endpoint: The API endpoint (e.g., 'dcim/sites', 'ipam/prefixes')
data: Object data to create
Returns:
The created object as a dict
Raises:
requests.HTTPError: If the request fails
"""
url = self._build_url(endpoint)
response = self.session.post(url, json=data, verify=self.verify_ssl)
response.raise_for_status()
return response.json()
def update(self, endpoint: str, id: int, data: Dict[str, Any]) -> Dict[str, Any]:
"""
Update an existing object in NetBox via the REST API.
Args:
endpoint: The API endpoint (e.g., 'dcim/sites', 'ipam/prefixes')
id: ID of the object to update
data: Object data to update
Returns:
The updated object as a dict
Raises:
requests.HTTPError: If the request fails
"""
url = self._build_url(endpoint, id)
response = self.session.patch(url, json=data, verify=self.verify_ssl)
response.raise_for_status()
return response.json()
def delete(self, endpoint: str, id: int) -> bool:
"""
Delete an object from NetBox via the REST API.
Args:
endpoint: The API endpoint (e.g., 'dcim/sites', 'ipam/prefixes')
id: ID of the object to delete
Returns:
True if deletion was successful, False otherwise
Raises:
requests.HTTPError: If the request fails
"""
url = self._build_url(endpoint, id)
response = self.session.delete(url, verify=self.verify_ssl)
response.raise_for_status()
return response.status_code == 204
def bulk_create(self, endpoint: str, data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Create multiple objects in NetBox via the REST API.
Args:
endpoint: The API endpoint (e.g., 'dcim/sites', 'ipam/prefixes')
data: List of object data to create
Returns:
List of created objects as dicts
Raises:
requests.HTTPError: If the request fails
"""
url = f"{self._build_url(endpoint)}bulk/"
response = self.session.post(url, json=data, verify=self.verify_ssl)
response.raise_for_status()
return response.json()
def bulk_update(self, endpoint: str, data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Update multiple objects in NetBox via the REST API.
Args:
endpoint: The API endpoint (e.g., 'dcim/sites', 'ipam/prefixes')
data: List of object data to update (must include ID)
Returns:
List of updated objects as dicts
Raises:
requests.HTTPError: If the request fails
"""
url = f"{self._build_url(endpoint)}bulk/"
response = self.session.patch(url, json=data, verify=self.verify_ssl)
response.raise_for_status()
return response.json()
def bulk_delete(self, endpoint: str, ids: List[int]) -> bool:
"""
Delete multiple objects from NetBox via the REST API.
Args:
endpoint: The API endpoint (e.g., 'dcim/sites', 'ipam/prefixes')
ids: List of IDs to delete
Returns:
True if deletion was successful, False otherwise
Raises:
requests.HTTPError: If the request fails
"""
url = f"{self._build_url(endpoint)}bulk/"
data = [{"id": id} for id in ids]
response = self.session.delete(url, json=data, verify=self.verify_ssl)
response.raise_for_status()
return response.status_code == 204