diff --git a/gso/oss-params-example.json b/gso/oss-params-example.json
index 1f25050be6547fc3241b819d5ee96db1e00eca0d..a06212ef7441a31055fdc0b240982aa9c2c00bf5 100644
--- a/gso/oss-params-example.json
+++ b/gso/oss-params-example.json
@@ -2,9 +2,9 @@
"GENERAL": {
"public_hostname": "https://gap.geant.org"
},
- "RESOURCE_MANAGEMENT": {
- "api": "https://127.0.0.1:8000"
- "token": "TOKEN"
+ "NETBOX": {
+ "api": "https://127.0.0.1:8000",
+ "token": "TOKEN"
},
"IPAM": {
"INFOBLOX": {
diff --git a/gso/services/netbox_client.py b/gso/services/netbox_client.py
index eb69515c6d593f4071b4fd0f8ea04b5f5066f581..84d9c7e5b8bdd3c7abb6bc992d6e4841c25b3918 100644
--- a/gso/services/netbox_client.py
+++ b/gso/services/netbox_client.py
@@ -3,14 +3,15 @@ This module contains all methods
to communicate with the
NetBox API endpoint: Data Center Infrastructure Main (dcim)
"""
-from typing import Optional
from uuid import UUID
-import pynetbox
import pydantic
-from pydantic import BaseModel
+import pynetbox
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
@@ -58,281 +59,188 @@ class Site(pydantic.BaseModel):
slug: str
-class ModuleInfo(BaseModel):
- device_type: str
- module_bays_slots: list[int]
- module_type: str
- breakout_interfaces_per_slot: list[int]
- total_10g_interfaces: int
-
-
-class TierInfo:
+class NetBoxClient:
def __init__(self):
- self.Tier1 = ModuleInfo(
- 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)
-
+ self.netbox = pynetbox.api(NetBoxParams.api, NetBoxParams.token)
-# An exception for not found search
-class NotFoundError(Exception):
- """Exception raised for not found search."""
- pass
+ def get_all_devices(self):
+ return list(self.netbox.dcim.devices.all())
+ 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
-class WorkflowStateException(Exception):
- """Exception raised on problems during workflow."""
- pass
+ def get_interfaces_by_device(self, device_name: str, speed: str):
+ """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)
+ )
-class NetBoxClient:
- token = "6762782659eba4eb2a490716093dba9f2fc31b36"
- api = "http://localhost:8000/"
-
- def connect(self):
+ def create_interface(self, iface_name: str, type: str, speed: str, device_name: str) -> dict:
"""
- Creates a netbox client to communicate
- with the NetBox endpoints
- The responses of the client are returned mostly as Records or RecordSet.
- For more details see the dcim.py file in the
- pynetbox package.
-
- 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/
+ 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 = self.get_device_by_name(device_name)
- return pynetbox.api(self.api, self.token)
-
-
-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)
+ new_iface = self.netbox.dcim.interfaces.create(
+ name=iface_name, type=type, speed=speed, enabled=False, mark_connected=False, device=device.id
+ )
- # Create device type
- device_type = nbclient.dcim.device_types.create(device_type)
- return dict(device_type)
+ return dict(new_iface)
+ 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:
- device_role = DeviceRole(**{"name": name, "slug": slug})
- device_role = dict(device_role)
- device_role = nbclient.dcim.device_roles.create(device_role)
- return dict(device_role)
+ # 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})
+ # Convert the format from DeviceType to dict
+ device_type = dict(device_type)
-def create_device_site(nbclient, name, slug) -> dict:
- device_site = Site(**{"name": name, "slug": slug})
- device_site = nbclient.dcim.sites.create(dict(device_site))
- return dict(device_site)
+ # Create device type
+ device_type = self.netbox.dcim.device_types.create(device_type)
+ return dict(device_type)
+ 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:
- device_manufacturer = Manufacturer(**{"name": name, "slug": slug})
- device_manufacturer = dict(device_manufacturer)
- device_manufacturer = nbclient.dcim.manufacturers.create(device_manufacturer)
- return dict(device_manufacturer)
+ def create_device_site(self, name, slug) -> dict:
+ device_site = Site(**{"name": name, "slug": slug})
+ device_site = self.netbox.dcim.sites.create(dict(device_site))
+ return dict(device_site)
+ 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:
- # fqdn: str, model: str, device_role: str, site: str
- nbclient = NetBoxClient().connect()
+ def create_device(self, router_name: str, site_tier: str) -> dict:
+ """Creates a new device in netbox."""
- # Get device type id
- tier_info = TierInfo().get_module_by_name(f"Tier{site_tier}")
- device_type = nbclient.dcim.device_types.get(model=tier_info.device_type)
+ # 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 = nbclient.dcim.device_roles.get(name="router")
+ # Get device role id
+ device_role = self.netbox.dcim.device_roles.get(name="router")
- # Get site id
- device_site = nbclient.dcim.sites.get(name="Amsterdam")
+ # Get site id
+ device_site = self.netbox.dcim.sites.get(name="Amsterdam")
- # Create new device
- device = nbclient.dcim.devices.create(
- 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",
+ # Create new device
+ device = self.netbox.dcim.devices.create(
+ name=router_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)
+ 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)
-
-
-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)
+ # Now get interface for device
+ iface = self.netbox.dcim.interfaces.get(name=iface_name, device_id=device.id)
- # Assign interface to lag
- iface.lag = lag.id
+ # Get lag
+ lag = self.netbox.dcim.interfaces.get(name=lag_name, device_id=device.id)
- # Update interface
- updated_iface = nbclient.dcim.interfaces.update(iface)
+ # Assign interface to lag
+ 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:
- # First get interface from device
- device = get_device_by_name(nbclient, device_name)
- interface = nbclient.dcim.interfaces.get(device_id=device.id,
- name=iface_name)
+ def reserve_interface(self, device_name: str, iface_name: str) -> dict:
+ """Reserve an interface by enabling it."""
- # Reserve interface by enabling it
- if interface is None:
- raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
+ # 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 is reserved
- if interface.enabled:
- raise WorkflowStateException(f"The interface: {iface_name} on device: {device_name} is already reserved.")
+ # Reserve interface by enabling it
+ if interface is None:
+ raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
- # Reserve interface by enabling it
- interface.enabled = True
- interface.save()
+ # Check if interface is reserved
+ if interface.enabled:
+ 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:
- # First get interface from device
- device = get_device_by_name(nbclient, device_name)
- interface = nbclient.dcim.interfaces.get(device_id=device.id,
- name=iface_name)
+ def allocate_interface(self, device_name: str, iface_name: str) -> dict:
+ """Allocate an interface by marking it as connected."""
- # allocate interface by marking it as connected
- # Check if interface is available
- if interface is None:
- raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
+ 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 reserved
- if interface.mark_connected:
- raise WorkflowStateException(f"The interface: {iface_name} on device: {device_name} is already allocated.")
+ # Check if interface is available
+ if interface is None:
+ raise NotFoundError(f"Interface: {iface_name} on device: {device_name} not found.")
- # allocate interface by mark as connected
- interface.mark_connected = True
- interface.save()
+ # Check if interface is reserved
+ if interface.mark_connected:
+ 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]:
- """Returns all available lags not assigned to a device."""
+ def get_available_lags(self, router_id: UUID) -> list[str]:
+ """Returns all available lags not assigned to a device."""
- nbclient = NetBoxClient().connect()
- router_name = Router.from_subscription(router_id).router.router_fqdn
- device = get_device_by_name(nbclient, router_name)
+ 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 nbclient.dcim.interfaces.filter(device_name=device.id, type="lag")
- ]
+ # Get the existing lag interfaces for the device
+ lag_interface_names = [
+ interface["name"] for interface in self.netbox.dcim.interfaces.filter(device_name=device.id, type="lag")
+ ]
- # Generate all feasible lags
- all_feasible_lags = [f"LAG-{i}" for i in FEASIBLE_LAG_RANGE]
+ # Generate all feasible lags
+ all_feasible_lags = [f"LAG-{i}" for i in FEASIBLE_LAG_RANGE]
- # Find available lags not assigned to the device
- available_lags = [lag for lag in all_feasible_lags if lag not in lag_interface_names]
- return available_lags
+ # Find available lags not assigned to the device
+ available_lags = [lag for lag in all_feasible_lags if lag not in lag_interface_names]
+ return available_lags
-if __name__ == "__main__":
- print(dict(create_device_manufacturer("Juniper", "juniper")))
+# if __name__ == "__main__":
+# print(dict(create_device_manufacturer("Juniper", "juniper")))
diff --git a/gso/settings.py b/gso/settings.py
index ddc7eaafb79900c4fd85e3afcd4bc5f4d36fdc07..f6ecc5e08e5cf9e45485b1931b2719661d56fc36 100644
--- a/gso/settings.py
+++ b/gso/settings.py
@@ -79,10 +79,11 @@ class ProvisioningProxyParams(BaseSettings):
api_version: int
-class ResourceManagementParams(BaseSettings):
- """TO DO: resource management parameters."""
+class NetBoxParams(BaseSettings):
+ """Parameters for NetBox."""
- todo: str
+ token: str
+ api: str
class OSSParams(BaseSettings):
@@ -90,7 +91,7 @@ class OSSParams(BaseSettings):
GENERAL: GeneralParams
IPAM: IPAMParams
- RESOURCE_MANAGEMENT: ResourceManagementParams
+ NETBOX: NetBoxParams
PROVISIONING_PROXY: ProvisioningProxyParams
diff --git a/gso/utils/__init__.py b/gso/utils/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391
diff --git a/gso/utils/device_info.py b/gso/utils/device_info.py
new file mode 100644
index 0000000000000000000000000000000000000000..47b2a8bed2d08e93debff83e55c28f6399dc2e5f
--- /dev/null
+++ b/gso/utils/device_info.py
@@ -0,0 +1,40 @@
+from typing import Optional
+
+from pydantic import BaseModel
+
+
+class ModuleInfo(BaseModel):
+ 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):
+ self.Tier1 = ModuleInfo(
+ 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)
diff --git a/gso/utils/exceptions.py b/gso/utils/exceptions.py
new file mode 100644
index 0000000000000000000000000000000000000000..a11a741fbd09c272a606d7927987ccf5baf7f058
--- /dev/null
+++ b/gso/utils/exceptions.py
@@ -0,0 +1,10 @@
+class NotFoundError(Exception):
+ """Exception raised for not found search."""
+
+ pass
+
+
+class WorkflowStateException(Exception):
+ """Exception raised on problems during workflow."""
+
+ pass
diff --git a/gso/workflows/router/create_router.py b/gso/workflows/router/create_router.py
index ba48e89e090e74949e0b246f3e68efb90f36a56d..403148d26995c813b2fa35790ea7d374219ffb9f 100644
--- a/gso/workflows/router/create_router.py
+++ b/gso/workflows/router/create_router.py
@@ -15,7 +15,8 @@ from gso.products.product_blocks.router import RouterRole, RouterVendor, generat
from gso.products.product_types.router import RouterInactive, RouterProvisioning
from gso.products.product_types.site import Site
from gso.products.shared import PortNumber
-from gso.services import infoblox, provisioning_proxy, subscriptions, netbox_client
+from gso.services import infoblox, provisioning_proxy, subscriptions
+from gso.services.netbox_client import NetBoxClient
from gso.services.provisioning_proxy import pp_interaction
from gso.workflows.utils import customer_selector, iso_from_ipv4
@@ -156,7 +157,7 @@ def provision_router_real(subscription: RouterProvisioning, process_id: UUIDstr,
@step("Create NetBox Device")
def create_netbox_device(subscription: RouterProvisioning) -> State:
- netbox_client.create_device(subscription.router.router_fqdn, subscription.router.router_site.site_tier)
+ NetBoxClient().create_device(subscription.router.router_fqdn, subscription.router.router_site.site_tier)
return {
"subscription": subscription
diff --git a/gso/workflows/router/terminate_router.py b/gso/workflows/router/terminate_router.py
index a564e48876ca06fedf90beeb6c4bbee875614fd7..b02f90cae1f328e148440ffc3f146d29cb1aa198 100644
--- a/gso/workflows/router/terminate_router.py
+++ b/gso/workflows/router/terminate_router.py
@@ -11,7 +11,7 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
from gso.products.product_types.router import Router
from gso.services import infoblox
-from gso.services import netbox_client
+from gso.services.netbox_client import NetBoxClient
logger = logging.getLogger(__name__)
@@ -61,7 +61,7 @@ def remove_config_from_router() -> None:
@step("Remove Device from NetBox")
def remove_device_from_netbox(subscription: Router) -> dict[str, Router]:
- netbox_client.delete_device(subscription.router.router_fqdn)
+ NetBoxClient().delete_device(subscription.router.router_fqdn)
return {"subscription": subscription}
diff --git a/utils/netboxcli.py b/utils/netboxcli.py
index 2ba7e2ed628db2f44286efa0ba0236eccd3dbc4f..2e5b680bd1b44fdf4ff8ec7b4a416aa8e0f40381 100644
--- a/utils/netboxcli.py
+++ b/utils/netboxcli.py
@@ -6,12 +6,7 @@ import click
import pandas as pd
from gso.services import netbox_client as nc
-
-# Define url and token for netbox API
-# TODO move to config file
-token = "428b68e7cb7afc6e7441879f21c3de6583081e40"
-api = 'http://127.0.0.1:8000'
-nbclient = nc.connect(api, token)
+from gso.services.netbox_client import NetBoxClient
def convert_to_table(data, fields):
@@ -41,7 +36,7 @@ def create():
@click.option('--site', default="Amsterdam", help='Device role')
def device(fqdn: str, model: str, role: str, site: str):
click.echo(f"Creating device: fqdn={fqdn}, model={model}, role={role}")
- new_device = nc.create_device(nbclient, fqdn, model, role, site)
+ new_device = NetBoxClient().create_device(fqdn, model)
print(new_device)
@@ -52,7 +47,7 @@ def device(fqdn: str, model: str, role: str, site: str):
@click.option('--fqdn', help='Device where to create interface')
def interface(name: str, type: str, speed: str, fqdn: str):
click.echo(f"Creating interface: name={name}, speed={speed}, fqdn={fqdn}")
- new_interface = nc.create_interface(nbclient, name, type, speed, fqdn)
+ new_interface = NetBoxClient().create_interface(name, type, speed, fqdn)
print(new_interface)
@@ -61,7 +56,7 @@ def interface(name: str, type: str, speed: str, fqdn: str):
@click.option('--slug', help='Short name for manufacturer')
def manufacturer(name: str, slug: str):
click.echo(f"Creating manufacturer: name={name}")
- manufacturer = nc.create_device_manufacturer(nbclient, name, slug)
+ manufacturer = NetBoxClient().create_device_manufacturer(name, slug)
print(manufacturer)
@@ -71,7 +66,7 @@ def manufacturer(name: str, slug: str):
@click.option('--slug', help='Short name for manufacturer')
def device_type(manufacturer: str, model: str, slug: str):
click.echo(f"Creating device type: manufacturer={manufacturer} model = {model}")
- device_type = nc.create_device_type(nbclient, manufacturer, model, slug)
+ device_type = NetBoxClient().create_device_type(manufacturer, model, slug)
print(device_type)
@@ -80,7 +75,7 @@ def device_type(manufacturer: str, model: str, slug: str):
@click.option('--slug', help='Short name for device role')
def device_role(name: str, slug: str):
click.echo(f"Creating device role: name={name}")
- device_role = nc.create_device_role(nbclient, name, slug)
+ device_role = NetBoxClient().create_device_role(name, slug)
print(device_role)
@@ -89,7 +84,7 @@ def device_role(name: str, slug: str):
@click.option('--slug', help='Short name for device site')
def device_site(name: str, slug: str):
click.echo(f"Creating device site: name={name}")
- device_site = nc.create_device_site(nbclient, name, slug)
+ device_site = NetBoxClient().create_device_site(name, slug)
print(device_site)
@@ -112,7 +107,7 @@ def list():
@click.option('--speed', default="1000", help='Interface speed to list interfaces (default 1000=1G)')
def interfaces(fqdn: str, speed: str):
click.echo(f"Listing all interfaces for: device with fqdn={fqdn}, speed={speed}")
- interface_list = nc.get_interfaces_by_device(nbclient, fqdn, speed)
+ interface_list = NetBoxClient().get_interfaces_by_device(fqdn, speed)
display_fields = ["name", "enabled", "mark_connected", "custom_fields", "lag", "speed"]
iface_list = []
for iface in interface_list:
@@ -125,7 +120,7 @@ def interfaces(fqdn: str, speed: str):
@list.command()
def devices():
click.echo("Listing all devices:")
- device_list = nc.get_all_devices(nbclient)
+ device_list = NetBoxClient().get_all_devices()
display_fields = ["name", "device_type"]
devices = []
for device in device_list:
@@ -151,7 +146,7 @@ def attach():
@click.option('--lag', help='LAG name to attach interface')
def interface(fqdn: str, iface: str, lag: str):
click.echo(f"Attaching interface to lag: device ={fqdn}, interface name={iface} to lag={lag}")
- new_iface = nc.attach_interface_to_lag(nbclient, fqdn, lag, iface)
+ new_iface = NetBoxClient().attach_interface_to_lag(fqdn, lag, iface)
print(new_iface)
@@ -169,7 +164,7 @@ def reserve():
@click.option('--iface', help='Interface name to reserve')
def interface(fqdn: str, iface: str):
click.echo(f"Reserving interface: device ={fqdn}, interface name={iface}")
- reserved_iface = nc.reserve_interface(nbclient, fqdn, iface)
+ reserved_iface = NetBoxClient().reserve_interface(fqdn, iface)
print(reserved_iface )