"""Contain all methods to communicate with the NetBox API endpoint. Data Center Infrastructure Main (DCIM)."""
from uuid import UUID
import pydantic
import pynetbox
from infoblox_client.objects import Interface
from pynetbox.models.dcim import Devices, DeviceTypes, Interfaces
from gso.products.product_types.router import Router
from gso.settings import load_oss_params
from gso.utils.device_info import DEFAULT_SITE, FEASIBLE_IP_TRUNK_LAG_RANGE, ROUTER_ROLE, TierInfo
from gso.utils.exceptions import NotFoundError, WorkflowStateError
class Manufacturer(pydantic.BaseModel):
"""Defines the manufacturer of a device."""
name: str
slug: str
class DeviceType(pydantic.BaseModel):
"""Defines the device type.
The manufacturer should be created first to get the manufacturer id, which is defined here as int.
"""
manufacturer: int
model: str
slug: str
class DeviceRole(pydantic.BaseModel):
"""Defines the role of a device."""
name: str
slug: str
class Site(pydantic.BaseModel):
"""Defines the site of a device."""
name: str
slug: str
class NetboxClient:
"""Implement all methods to communicate with the Netbox :term:`API`."""
def __init__(self) -> None:
self.netbox_params = load_oss_params().NETBOX
self.netbox = pynetbox.api(self.netbox_params.api, self.netbox_params.token)
def get_all_devices(self) -> list[Devices]:
return list(self.netbox.dcim.devices.all())
def get_allocated_interfaces_by_gso_subscription(self, device_name: str, subscription_id: UUID) -> list[Interfaces]:
"""Return all allocated interfaces of a device by name."""
device = self.get_device_by_name(device_name)
return self.netbox.dcim.interfaces.filter(
device_id=device.id, enabled=True, mark_connected=True, description=subscription_id
)
def get_device_by_name(self, device_name: str) -> Devices:
"""Return the device object by name from netbox, or ``None`` if not found."""
return self.netbox.dcim.devices.get(name=device_name)
def get_interfaces_by_device(self, device_name: str, speed: str) -> list[Interfaces]:
"""Get all interfaces of a device by name and speed."""
device = self.get_device_by_name(device_name)
return list(
self.netbox.dcim.interfaces.filter(device_id=device.id, enabled=False, mark_connected=False, speed=speed)
)
def create_interface(
self, iface_name: str, type: str, device_name: str, description: str | None = None, enabled: bool = False
) -> Interfaces:
"""Create new interface on a device, where device is defined by name.
The type parameter can be 1000base-t, 10gbase-t, lag, etc.
For more details on type definition have a look in choices.py in the netbox API implementation in module DCIM.
Returns the new interface object as dict.
"""
device = self.get_device_by_name(device_name)
return self.netbox.dcim.interfaces.create(
name=iface_name,
type=type,
enabled=enabled,
mark_connected=False,
device=device.id,
description=description,
)
def delete_interface(self, device_name: str, iface_name: str) -> None:
"""Delete an interface from a device by name."""
device = self.get_device_by_name(device_name)
interface = self.netbox.dcim.interfaces.get(device_id=device.id, name=iface_name)
return interface.delete()
def create_device_type(self, manufacturer: str, model: str, slug: str) -> DeviceTypes:
"""Create a new device type in Netbox."""
# First get manufacturer id
manufacturer_id = int(self.netbox.dcim.manufacturers.get(name=manufacturer).id)
device_type = DeviceType(
**{"manufacturer": manufacturer_id, "model": model, "slug": slug} # type: ignore[arg-type]
)
return self.netbox.dcim.device_types.create(dict(device_type))
def create_device_role(self, name: str, slug: str) -> DeviceRole:
device_role = DeviceRole(**{"name": name, "slug": slug})
return self.netbox.dcim.device_roles.create(dict(device_role))
def create_device_site(self, name: str, slug: str) -> Site:
device_site = Site(**{"name": name, "slug": slug})
return self.netbox.dcim.sites.create(dict(device_site))
def create_device_manufacturer(self, name: str, slug: str) -> Manufacturer:
device_manufacturer = Manufacturer(**{"name": name, "slug": slug})
return self.netbox.dcim.manufacturers.create(dict(device_manufacturer))
@staticmethod
def calculate_interface_speed(interface: Interfaces) -> int | None:
"""Calculate the interface speed in bits per second."""
type_parts = interface.type.value.split("-")
if "gbase" in type_parts[0]:
return int("".join(filter(str.isdigit, type_parts[0]))) * 1000000
return None
def create_device(self, device_name: str, site_tier: str) -> Devices:
"""Create a new device in Netbox."""
# Get device type id
tier_info = TierInfo().get_module_by_name(f"Tier{site_tier}")
device_type = self.netbox.dcim.device_types.get(model=tier_info.device_type)
# Get device role id
device_role = self.netbox.dcim.device_roles.get(name=ROUTER_ROLE["name"])
# Get site id
device_site = self.netbox.dcim.sites.get(name=DEFAULT_SITE["name"])
# Create new device
device = self.netbox.dcim.devices.create(
name=device_name, device_type=device_type.id, role=device_role.id, site=device_site.id
)
module_bays = list(self.netbox.dcim.module_bays.filter(device_id=device.id))
card_type = self.netbox.dcim.module_types.get(model=tier_info.module_type)
valid_module_bays = [bay for bay in module_bays if int(bay.position) in tier_info.module_bays_slots]
for module_bay in valid_module_bays:
self.netbox.dcim.modules.create(
device=device.id,
module_bay=module_bay.id,
module_type=card_type.id,
status="active",
enabled=False,
comments="Installed via pynetbox",
)
for interface in self.netbox.dcim.interfaces.filter(device_id=device.id):
interface.speed = self.calculate_interface_speed(interface)
interface.enabled = False
interface.save()
return device
def delete_device(self, device_name: str) -> None:
"""Delete device by name"""
self.netbox.dcim.devices.get(name=device_name).delete()
return
def attach_interface_to_lag(
self, device_name: str, lag_name: str, iface_name: str, description: str | None = None
) -> Interfaces:
"""Assign a given interface to a :term:`LAG`.
Returns the interface object after assignment.
"""
# Get device id
device = self.get_device_by_name(device_name)
# Get interface for device
iface = self.netbox.dcim.interfaces.get(name=iface_name, device_id=device.id)
# Get LAG
lag = self.netbox.dcim.interfaces.get(name=lag_name, device_id=device.id)
# Assign interface to LAG, ensuring it doesn't already belong to a LAG
if iface.lag:
raise WorkflowStateError(
f"The interface: {iface_name} on device: {device_name} already belongs to a LAG: {iface.lag.name}."
)
iface.lag = lag.id
# Set description if provided
if description:
iface.description = description
iface.save()
return iface
def reserve_interface(self, device_name: str, iface_name: str) -> Interfaces:
"""Reserve an interface by enabling it."""
# First get interface from device
device = self.get_device_by_name(device_name)
interface = self.netbox.dcim.interfaces.get(device_id=device.id, name=iface_name)
# Check if interface exists
if interface is None:
raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
# Check if interface is reserved
if interface.enabled:
raise WorkflowStateError(f"The interface: {iface_name} on device: {device_name} is already reserved.")
# Reserve interface by enabling it
interface.enabled = True
interface.save()
return interface
def allocate_interface(self, device_name: str, iface_name: str) -> Interfaces:
"""Allocate an interface by marking it as connected."""
device = self.get_device_by_name(device_name)
interface = self.netbox.dcim.interfaces.get(device_id=device.id, name=iface_name)
# Check if interface exists
if interface is None:
raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
# Check if interface is reserved
if interface.mark_connected:
raise WorkflowStateError(f"The interface: {iface_name} on device: {device_name} is already allocated.")
# Allocate interface by marking it as connected
interface.mark_connected = True
interface.save()
return interface
def free_interface(self, device_name: str, iface_name: str) -> Interfaces:
"""Free interface by marking disconnect and disable it."""
device = self.get_device_by_name(device_name)
interface = self.netbox.dcim.interfaces.get(device_id=device.id, name=iface_name)
# Check if interface is available
if interface is None:
raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
interface.mark_connected = False
interface.enabled = False
interface.description = ""
interface.save()
return interface
def detach_interfaces_from_lag(self, device_name: str, lag_name: str) -> None:
"""Detach all interfaces from a LAG."""
device = self.get_device_by_name(device_name)
lag = self.netbox.dcim.interfaces.get(device_id=device.id, name=lag_name)
for interface in self.netbox.dcim.interfaces.filter(
device_id=device.id, lag_id=lag.id, enabled=False, mark_connected=False
):
interface.lag = None
interface.save()
return
def get_available_lags(self, router_id: UUID) -> list[str]:
"""Return all available :term:`LAG`s not assigned to a device."""
router_name = Router.from_subscription(router_id).router.router_fqdn
device = self.get_device_by_name(router_name)
# Get the existing LAG interfaces for the device
lag_interface_names = [
interface["name"] for interface in self.netbox.dcim.interfaces.filter(device=device.name, type="lag")
]
# Generate all feasible LAGs
all_feasible_lags = [f"LAG-{i}" for i in FEASIBLE_IP_TRUNK_LAG_RANGE]
# Return available LAGs not assigned to the device
return [lag for lag in all_feasible_lags if lag not in lag_interface_names]
@staticmethod
def calculate_speed_bits_per_sec(speed: str) -> int:
"""Extract the numeric part from the speed."""
numeric_part = int("".join(filter(str.isdigit, speed)))
# Convert to bits per second
return numeric_part * 1000000
def get_available_interfaces(self, router_id: UUID, speed: str) -> Interfaces:
"""Return all available interfaces of a device filtered by speed."""
router = Router.from_subscription(router_id).router.router_fqdn
device = self.get_device_by_name(router)
speed_bps = self.calculate_speed_bits_per_sec(speed)
return self.netbox.dcim.interfaces.filter(
device=device.name, enabled=False, mark_connected=False, speed=speed_bps
)