Skip to content
Snippets Groups Projects

Netbox integration including intial CLI for populating base data and ...

Merged Netbox integration including intial CLI for populating base data and ...
netbox-integration into develop
Merged Neda Moeini requested to merge netbox-integration into develop
Viewing commit f2faa18c
Prev Next
Show latest version
9 files
+ 216
− 261
Compare changes
  • Side-by-side
  • Inline
Files
9
gso/services/netbox_client.py
+ 142
− 234
@@ -3,14 +3,15 @@ This module contains all methods
@@ -3,14 +3,15 @@ This module contains all methods
to communicate with the
to communicate with the
NetBox API endpoint: Data Center Infrastructure Main (dcim)
NetBox API endpoint: Data Center Infrastructure Main (dcim)
"""
"""
from typing import Optional
from uuid import UUID
from uuid import UUID
import pynetbox
import pydantic
import pydantic
from pydantic import BaseModel
import pynetbox
from gso.products import Router
from gso.products import Router
 
from gso.settings import NetBoxParams
 
from gso.utils.device_info import TierInfo, FEASIBLE_LAG_RANGE
 
from gso.utils.exceptions import NotFoundError, WorkflowStateException
# Define device models
# Define device models
@@ -58,281 +59,188 @@ class Site(pydantic.BaseModel):
@@ -58,281 +59,188 @@ class Site(pydantic.BaseModel):
slug: str
slug: str
class ModuleInfo(BaseModel):
class NetBoxClient:
device_type: str
module_bays_slots: list[int]
module_type: str
breakout_interfaces_per_slot: list[int]
total_10g_interfaces: int
class TierInfo:
def __init__(self):
def __init__(self):
self.Tier1 = ModuleInfo(
self.netbox = pynetbox.api(NetBoxParams.api, NetBoxParams.token)
device_type="7750-SR7s",
module_bays_slots=[1, 2],
module_type="XCM2s-XMA2s-36p-800g",
breakout_interfaces_per_slot=[36, 35, 34, 33],
total_10g_interfaces=80,
)
self.Tier2 = ModuleInfo(
device_type="7750-SR7s",
module_bays_slots=[1, 2],
module_type="XCM2s-XMA2s-36p-400g",
breakout_interfaces_per_slot=[36, 35, 34, 33],
total_10g_interfaces=60,
)
def get_module_by_name(self, name: str) -> Optional[ModuleInfo]:
if name == "Tier1":
return self.Tier1
elif name == "Tier2":
return self.Tier2
else:
return None
FEASIBLE_LAG_RANGE = range(1, 11)
# An exception for not found search
def get_all_devices(self):
class NotFoundError(Exception):
return list(self.netbox.dcim.devices.all())
"""Exception raised for not found search."""
pass
 
def get_device_by_name(self, device_name: str):
 
"""Returns the device object by name from netbox, or None if not found."""
 
return self.netbox.dcim.devices.get(name=device_name)
# An exception on a workflow error
def get_interfaces_by_device(self, device_name: str, speed: str):
class WorkflowStateException(Exception):
"""Get all interfaces of a device by name and speed."""
"""Exception raised on problems during workflow."""
pass
 
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)
 
)
class NetBoxClient:
def create_interface(self, iface_name: str, type: str, speed: str, device_name: str) -> dict:
token = "6762782659eba4eb2a490716093dba9f2fc31b36"
api = "http://localhost:8000/"
def connect(self):
"""
"""
Creates a netbox client to communicate
Creates a new interface on
with the NetBox endpoints
a device, where device is defined by name.
The responses of the client are returned mostly as Records or RecordSet.
The type parameter can be 1000base-t, 10gbase-t, lag...
For more details see the dcim.py file in the
For more details on type definidtion have a look in
pynetbox package.
choises.py in the netbox API implementation in module dcim.
Returns the new interface object as dict.
To create the client a main URL to NetBox is required. Something like:
api = 'http://127.0.0.1:8001'
To create a token, login to the NetBox web frontend or
post request directly a token from the NetBox API:
http://127.0.0.1:8001/api/users/tokens/provision/
"""
"""
 
device = self.get_device_by_name(device_name)
return pynetbox.api(self.api, self.token)
new_iface = self.netbox.dcim.interfaces.create(
name=iface_name, type=type, speed=speed, enabled=False, mark_connected=False, device=device.id
)
def get_all_devices(nbclient):
return list(nbclient.dcim.devices.all())
def get_device_by_name(nbclient, device_name: str) -> dict | None:
""" Returns the device object by name from netbox, or None if not found."""
return nbclient.dcim.devices.get(name=device_name)
# get all interfaces for a device
def get_interfaces_by_device(nbclient, device_name: str, speed: str):
device = get_device_by_name(nbclient, device_name)
return list(nbclient.dcim.interfaces.filter(device_id=device.id,
enabled=False,
mark_connected=False,
speed=speed
))
# Create a interface
def create_interface(nbclient,
iface_name: str,
type: str,
speed: str,
device_name: str) -> dict:
"""
Creates a new interface on
a device, where device is defined by name.
The type parameter can be 1000base-t, 10gbase-t, lag...
For more details on type definidtion have a look in
choises.py in the netbox API implementation in module dcim.
Returns the new interface object as dict.
"""
device = get_device_by_name(nbclient, device_name)
new_iface = nbclient.dcim.interfaces.create(name=iface_name,
type=type,
speed=speed,
enabled=False,
mark_connected=False,
device=device.id)
return dict(new_iface)
def create_device_type(nbclient, manufacturer: str, model: str, slug: str) -> dict:
# First get manufacturer id
manufacturer_id = int(nbclient.dcim.manufacturers.get(name=manufacturer).id)
device_type = DeviceType(**{"manufacturer": manufacturer_id, "model": model, "slug": slug})
# Convert the format from DeviceType to dict
device_type = dict(device_type)
# Create device type
return dict(new_iface)
device_type = nbclient.dcim.device_types.create(device_type)
return dict(device_type)
 
def create_device_type(self, manufacturer: str, model: str, slug: str) -> dict:
 
"""Creates a new device type in netbox."""
def create_device_role(nbclient, name: str, slug: str) -> dict:
# First get manufacturer id
device_role = DeviceRole(**{"name": name, "slug": slug})
manufacturer_id = int(self.netbox.dcim.manufacturers.get(name=manufacturer).id)
device_role = dict(device_role)
device_type = DeviceType(**{"manufacturer": manufacturer_id, "model": model, "slug": slug})
device_role = nbclient.dcim.device_roles.create(device_role)
return dict(device_role)
 
# Convert the format from DeviceType to dict
 
device_type = dict(device_type)
def create_device_site(nbclient, name, slug) -> dict:
# Create device type
device_site = Site(**{"name": name, "slug": slug})
device_type = self.netbox.dcim.device_types.create(device_type)
device_site = nbclient.dcim.sites.create(dict(device_site))
return dict(device_type)
return dict(device_site)
 
def create_device_role(self, name: str, slug: str) -> dict:
 
device_role = DeviceRole(**{"name": name, "slug": slug})
 
device_role = dict(device_role)
 
device_role = self.netbox.dcim.device_roles.create(device_role)
 
return dict(device_role)
def create_device_manufacturer(nbclient, name: str, slug: str) -> dict:
def create_device_site(self, name, slug) -> dict:
device_manufacturer = Manufacturer(**{"name": name, "slug": slug})
device_site = Site(**{"name": name, "slug": slug})
device_manufacturer = dict(device_manufacturer)
device_site = self.netbox.dcim.sites.create(dict(device_site))
device_manufacturer = nbclient.dcim.manufacturers.create(device_manufacturer)
return dict(device_site)
return dict(device_manufacturer)
 
def create_device_manufacturer(self, name: str, slug: str) -> dict:
 
device_manufacturer = Manufacturer(**{"name": name, "slug": slug})
 
device_manufacturer = dict(device_manufacturer)
 
device_manufacturer = self.netbox.dcim.manufacturers.create(device_manufacturer)
 
return dict(device_manufacturer)
def create_device(router_name: str, site_tier: str) -> dict:
def create_device(self, router_name: str, site_tier: str) -> dict:
# fqdn: str, model: str, device_role: str, site: str
"""Creates a new device in netbox."""
nbclient = NetBoxClient().connect()
# Get device type id
# Get device type id
tier_info = TierInfo().get_module_by_name(f"Tier{site_tier}")
tier_info = TierInfo().get_module_by_name(f"Tier{site_tier}")
device_type = nbclient.dcim.device_types.get(model=tier_info.device_type)
device_type = self.netbox.dcim.device_types.get(model=tier_info.device_type)
# Get device role id
# Get device role id
device_role = nbclient.dcim.device_roles.get(name="router")
device_role = self.netbox.dcim.device_roles.get(name="router")
# Get site id
# Get site id
device_site = nbclient.dcim.sites.get(name="Amsterdam")
device_site = self.netbox.dcim.sites.get(name="Amsterdam")
# Create new device
# Create new device
device = nbclient.dcim.devices.create(
device = self.netbox.dcim.devices.create(
name=router_name, device_type=device_type.id, role=device_role.id, site=device_site.id
name=router_name, device_type=device_type.id, role=device_role.id, site=device_site.id
)
module_bays = list(nbclient.dcim.module_bays.filter(device_id=device.id))
card_type = nbclient.dcim.module_types.get(model=tier_info.module_type)
for module_bay in module_bays:
nbclient.dcim.modules.create(
device=device.id,
module_bay=module_bay.id,
module_type=card_type.id,
status="active",
comments="Installed via pynetbox",
)
)
 
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)
 
for module_bay in module_bays:
 
self.netbox.dcim.modules.create(
 
device=device.id,
 
module_bay=module_bay.id,
 
module_type=card_type.id,
 
status="active",
 
comments="Installed via pynetbox",
 
)
 
 
return dict(device)
 
 
def delete_device(self, router_name: str):
 
self.netbox.dcim.devices.get(name=router_name).delete()
 
return
 
 
def attach_interface_to_lag(self, device_name: str, lag_name: str, iface_name: str) -> dict:
 
"""
 
Assign a given interface to a lag.
 
Returns the lag object with the assignend interfaces
 
"""
 
# Get device id
 
device = self.get_device_by_name(device_name)
return dict(device)
# Now get interface for device
iface = self.netbox.dcim.interfaces.get(name=iface_name, device_id=device.id)
def delete_device(router_name: str):
nbclient = NetBoxClient().connect()
nbclient.dcim.devices.get(name=router_name).delete()
return
def attach_interface_to_lag(nbclient, device_name: str, lag_name: str, iface_name: str) -> dict:
"""
Assign a given interface to a lag.
Returns the lag object with the assignend interfaces
"""
# Get device id
device = get_device_by_name(nbclient, device_name)
# Now get interface for device
iface = nbclient.dcim.interfaces.get(name=iface_name, device_id=device.id)
# Get lag
lag = nbclient.dcim.interfaces.get(name=lag_name, device_id=device.id)
# Assign interface to lag
# Get lag
iface.lag = lag.id
lag = self.netbox.dcim.interfaces.get(name=lag_name, device_id=device.id)
# Update interface
# Assign interface to lag
updated_iface = nbclient.dcim.interfaces.update(iface)
iface.lag = lag.id
return dict(updated_iface)
# Update interface
 
updated_iface = self.netbox.dcim.interfaces.update(iface)
 
return dict(updated_iface)
def reserve_interface(nbclient, device_name: str, iface_name: str) -> dict:
def reserve_interface(self, device_name: str, iface_name: str) -> dict:
# First get interface from device
"""Reserve an interface by enabling it."""
device = get_device_by_name(nbclient, device_name)
interface = nbclient.dcim.interfaces.get(device_id=device.id,
name=iface_name)
# Reserve interface by enabling it
# First get interface from device
if interface is None:
device = self.get_device_by_name(device_name)
raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
interface = self.netbox.dcim.interfaces.get(device_id=device.id, name=iface_name)
# Check if interface is reserved
# Reserve interface by enabling it
if interface.enabled:
if interface is None:
raise WorkflowStateException(f"The interface: {iface_name} on device: {device_name} is already reserved.")
raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
# Reserve interface by enabling it
# Check if interface is reserved
interface.enabled = True
if interface.enabled:
interface.save()
raise WorkflowStateException(f"The interface: {iface_name} on device: {device_name} is already reserved.")
return dict(interface)
# Reserve interface by enabling it
 
interface.enabled = True
 
interface.save()
 
return dict(interface)
def allocate_interface(nbclient, device_name: str, iface_name: str) -> dict:
def allocate_interface(self, device_name: str, iface_name: str) -> dict:
# First get interface from device
"""Allocate an interface by marking it as connected."""
device = get_device_by_name(nbclient, device_name)
interface = nbclient.dcim.interfaces.get(device_id=device.id,
name=iface_name)
# allocate interface by marking it as connected
device = self.get_device_by_name(device_name)
# Check if interface is available
interface = self.netbox.dcim.interfaces.get(device_id=device.id, name=iface_name)
if interface is None:
raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
# Check if interface is reserved
# Check if interface is available
if interface.mark_connected:
if interface is None:
raise WorkflowStateException(f"The interface: {iface_name} on device: {device_name} is already allocated.")
raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
# allocate interface by mark as connected
# Check if interface is reserved
interface.mark_connected = True
if interface.mark_connected:
interface.save()
raise WorkflowStateException(f"The interface: {iface_name} on device: {device_name} is already allocated.")
return dict(interface)
# allocate interface by mark as connected
 
interface.mark_connected = True
 
interface.save()
 
return dict(interface)
def get_available_lags(router_id: UUID) -> list[str]:
def get_available_lags(self, router_id: UUID) -> list[str]:
"""Returns all available lags not assigned to a device."""
"""Returns all available lags not assigned to a device."""
nbclient = NetBoxClient().connect()
router_name = Router.from_subscription(router_id).router.router_fqdn
router_name = Router.from_subscription(router_id).router.router_fqdn
device = self.get_device_by_name(router_name)
device = get_device_by_name(nbclient, router_name)
# Get the existing lag interfaces for the device
# Get the existing lag interfaces for the device
lag_interface_names = [
lag_interface_names = [
interface["name"] for interface in nbclient.dcim.interfaces.filter(device_name=device.id, type="lag")
interface["name"] for interface in self.netbox.dcim.interfaces.filter(device_name=device.id, type="lag")
]
]
# Generate all feasible lags
# Generate all feasible lags
all_feasible_lags = [f"LAG-{i}" for i in FEASIBLE_LAG_RANGE]
all_feasible_lags = [f"LAG-{i}" for i in FEASIBLE_LAG_RANGE]
# Find available lags not assigned to the device
# Find available lags not assigned to the device
available_lags = [lag for lag in all_feasible_lags if lag not in lag_interface_names]
available_lags = [lag for lag in all_feasible_lags if lag not in lag_interface_names]
return available_lags
return available_lags
if __name__ == "__main__":
# if __name__ == "__main__":
print(dict(create_device_manufacturer("Juniper", "juniper")))
# print(dict(create_device_manufacturer("Juniper", "juniper")))