Skip to content

Netbox client

Contain all methods to communicate with the NetBox API endpoint. Data Center Infrastructure Main (DCIM).

Manufacturer

Bases: BaseModel

Defines the manufacturer of a device.

Source code in gso/services/netbox_client.py 
23
24
25
26
27
class Manufacturer(pydantic.BaseModel):
    """Defines the manufacturer of a device."""

    name: str
    slug: str

DeviceType

Bases: BaseModel

Defines the device type.

The manufacturer should be created first to get the manufacturer id, which is defined here as int.

Source code in gso/services/netbox_client.py 
30
31
32
33
34
35
36
37
38
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

DeviceRole

Bases: BaseModel

Defines the role of a device.

Source code in gso/services/netbox_client.py 
41
42
43
44
45
class DeviceRole(pydantic.BaseModel):
    """Defines the role of a device."""

    name: str
    slug: str

Site

Bases: BaseModel

Defines the site of a device.

Source code in gso/services/netbox_client.py 
48
49
50
51
52
class Site(pydantic.BaseModel):
    """Defines the site of a device."""

    name: str
    slug: str

NetboxClient

Implement all methods to communicate with the Netbox API.

Source code in gso/services/netbox_client.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
class NetboxClient:
    """Implement all methods to communicate with the Netbox API."""

    def __init__(self) -> None:
        """Instantiate a new Netbox client."""
        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]:
        """Get all devices in Netbox."""
        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 raise not found."""
        device = self.netbox.dcim.devices.get(name=device_name)
        if device is None:
            msg = f"Device: {device_name} not found."
            raise NotFoundError(msg)
        return device

    def get_interface_by_name_and_device(self, iface_name: str, device_name: str) -> Interfaces:
        """Return the interface lists by name and device name from netbox."""
        device = self.get_device_by_name(device_name)
        interface = self.netbox.dcim.interfaces.get(device_id=device.id, name=iface_name)
        if interface is None:
            msg = f"Interface: {iface_name} on device with id: {device.id} not found."
            raise NotFoundError(msg)
        return interface

    def get_interfaces_by_device(self, device_name: str, speed: str) -> list[Interfaces]:
        """Get all interfaces of a device by name and speed that are not reserved and not allocated."""
        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,
        interface_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=interface_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."""
        interface = self.get_interface_by_name_and_device(iface_name, device_name)
        if interface:
            return interface.delete()
        return None

    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)
        return self.netbox.dcim.device_types.create(dict(device_type))

    def create_device_role(self, name: str, slug: str) -> DeviceRole:
        """Create a new device role."""
        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:
        """Create a new site for devices."""
        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:
        """Create a new device 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 if exists."""
        with suppress(AttributeError):
            self.netbox.dcim.devices.get(name=device_name).delete()

    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 LAG.

        Returns:
        the interface object after assignment.
        """
        iface = self.get_interface_by_name_and_device(iface_name, device_name)

        # Get LAG
        lag = self.get_interface_by_name_and_device(lag_name, device_name)

        # Assign interface to LAG, ensuring it does not already belong to a LAG.
        if iface.lag:
            msg = f"The interface: {iface_name} on device: {device_name} already belongs to a LAG: {iface.lag.name}."
            raise WorkflowStateError(msg)
        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
        interface = self.get_interface_by_name_and_device(iface_name, device_name)

        # Check if interface is reserved
        if interface.enabled:
            msg = f"The interface: {iface_name} on device: {device_name} is already reserved."
            raise WorkflowStateError(msg)

        # 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."""
        # First get interface from device
        interface = self.get_interface_by_name_and_device(iface_name, device_name)

        # Check if interface is reserved
        if interface.mark_connected:
            msg = f"The interface: {iface_name} on device: {device_name} is already allocated."
            raise WorkflowStateError(msg)

        # 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."""
        # First get interface from device
        interface = self.get_interface_by_name_and_device(iface_name, device_name)
        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()

    def get_available_lags_in_range(self, router_id: UUID, lag_range: range) -> list[str]:
        """Return all available LAGs within a given range 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 in the specified range
        all_feasible_lags = [f"lag-{i}" for i in 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]

    def get_available_lags(self, router_id: UUID) -> list[str]:
        """Return all available LAG not assigned to a device."""
        return self.get_available_lags_in_range(router_id, FEASIBLE_IP_TRUNK_LAG_RANGE)

    def get_available_services_lags(self, router_id: UUID) -> list[str]:
        """Return all available Edge port LAGs not assigned to a device."""
        return self.get_available_lags_in_range(router_id, FEASIBLE_SERVICES_LAG_RANGE)

    @staticmethod
    def calculate_speed_bits_per_sec(speed: str) -> int:
        """Calculate 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 | UUIDstr, 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,
        )

__init__()

Instantiate a new Netbox client.

Source code in gso/services/netbox_client.py 
58
59
60
61
def __init__(self) -> None:
    """Instantiate a new Netbox client."""
    self.netbox_params = load_oss_params().NETBOX
    self.netbox = pynetbox.api(self.netbox_params.api, self.netbox_params.token)

get_all_devices()

Get all devices in Netbox.

Source code in gso/services/netbox_client.py 
63
64
65
def get_all_devices(self) -> list[Devices]:
    """Get all devices in Netbox."""
    return list(self.netbox.dcim.devices.all())

get_allocated_interfaces_by_gso_subscription(device_name, subscription_id)

Return all allocated interfaces of a device by name.

Source code in gso/services/netbox_client.py 
67
68
69
70
71
72
73
74
75
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,
    )

get_device_by_name(device_name)

Return the device object by name from netbox, or raise not found.

Source code in gso/services/netbox_client.py 
77
78
79
80
81
82
83
def get_device_by_name(self, device_name: str) -> Devices:
    """Return the device object by name from netbox, or raise not found."""
    device = self.netbox.dcim.devices.get(name=device_name)
    if device is None:
        msg = f"Device: {device_name} not found."
        raise NotFoundError(msg)
    return device

get_interface_by_name_and_device(iface_name, device_name)

Return the interface lists by name and device name from netbox.

Source code in gso/services/netbox_client.py 
85
86
87
88
89
90
91
92
def get_interface_by_name_and_device(self, iface_name: str, device_name: str) -> Interfaces:
    """Return the interface lists by name and device name from netbox."""
    device = self.get_device_by_name(device_name)
    interface = self.netbox.dcim.interfaces.get(device_id=device.id, name=iface_name)
    if interface is None:
        msg = f"Interface: {iface_name} on device with id: {device.id} not found."
        raise NotFoundError(msg)
    return interface

get_interfaces_by_device(device_name, speed)

Get all interfaces of a device by name and speed that are not reserved and not allocated.

Source code in gso/services/netbox_client.py 
94
95
96
97
98
99
def get_interfaces_by_device(self, device_name: str, speed: str) -> list[Interfaces]:
    """Get all interfaces of a device by name and speed that are not reserved and not allocated."""
    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),
    )

create_interface(iface_name, interface_type, device_name, description=None, *, enabled=False)

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.

Source code in gso/services/netbox_client.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def create_interface(
    self,
    iface_name: str,
    interface_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=interface_type,
        enabled=enabled,
        mark_connected=False,
        device=device.id,
        description=description,
    )

delete_interface(device_name, iface_name)

Delete an interface from a device by name.

Source code in gso/services/netbox_client.py 
129
130
131
132
133
134
def delete_interface(self, device_name: str, iface_name: str) -> None:
    """Delete an interface from a device by name."""
    interface = self.get_interface_by_name_and_device(iface_name, device_name)
    if interface:
        return interface.delete()
    return None

create_device_type(manufacturer, model, slug)

Create a new device type in Netbox.

Source code in gso/services/netbox_client.py 
136
137
138
139
140
141
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)
    return self.netbox.dcim.device_types.create(dict(device_type))

create_device_role(name, slug)

Create a new device role.

Source code in gso/services/netbox_client.py 
143
144
145
146
def create_device_role(self, name: str, slug: str) -> DeviceRole:
    """Create a new device role."""
    device_role = DeviceRole(name=name, slug=slug)
    return self.netbox.dcim.device_roles.create(dict(device_role))

create_device_site(name, slug)

Create a new site for devices.

Source code in gso/services/netbox_client.py 
148
149
150
151
def create_device_site(self, name: str, slug: str) -> Site:
    """Create a new site for devices."""
    device_site = Site(name=name, slug=slug)
    return self.netbox.dcim.sites.create(dict(device_site))

create_device_manufacturer(name, slug)

Create a new device manufacturer.

Source code in gso/services/netbox_client.py 
153
154
155
156
def create_device_manufacturer(self, name: str, slug: str) -> Manufacturer:
    """Create a new device manufacturer."""
    device_manufacturer = Manufacturer(name=name, slug=slug)
    return self.netbox.dcim.manufacturers.create(dict(device_manufacturer))

calculate_interface_speed(interface) staticmethod

Calculate the interface speed in bits per second.

Source code in gso/services/netbox_client.py 
158
159
160
161
162
163
164
@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

create_device(device_name, site_tier)

Create a new device in Netbox.

Source code in gso/services/netbox_client.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
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

delete_device(device_name)

Delete device by name if exists.

Source code in gso/services/netbox_client.py 
205
206
207
208
def delete_device(self, device_name: str) -> None:
    """Delete device by name if exists."""
    with suppress(AttributeError):
        self.netbox.dcim.devices.get(name=device_name).delete()

attach_interface_to_lag(device_name, lag_name, iface_name, description=None)

Assign a given interface to a LAG.

Returns: the interface object after assignment.

Source code in gso/services/netbox_client.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
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 LAG.

    Returns:
    the interface object after assignment.
    """
    iface = self.get_interface_by_name_and_device(iface_name, device_name)

    # Get LAG
    lag = self.get_interface_by_name_and_device(lag_name, device_name)

    # Assign interface to LAG, ensuring it does not already belong to a LAG.
    if iface.lag:
        msg = f"The interface: {iface_name} on device: {device_name} already belongs to a LAG: {iface.lag.name}."
        raise WorkflowStateError(msg)
    iface.lag = lag.id

    # Set description if provided
    if description:
        iface.description = description

    iface.save()
    return iface

reserve_interface(device_name, iface_name)

Reserve an interface by enabling it.

Source code in gso/services/netbox_client.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
def reserve_interface(self, device_name: str, iface_name: str) -> Interfaces:
    """Reserve an interface by enabling it."""
    # First get interface from device
    interface = self.get_interface_by_name_and_device(iface_name, device_name)

    # Check if interface is reserved
    if interface.enabled:
        msg = f"The interface: {iface_name} on device: {device_name} is already reserved."
        raise WorkflowStateError(msg)

    # Reserve interface by enabling it
    interface.enabled = True
    interface.save()

    return interface

allocate_interface(device_name, iface_name)

Allocate an interface by marking it as connected.

Source code in gso/services/netbox_client.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def allocate_interface(self, device_name: str, iface_name: str) -> Interfaces:
    """Allocate an interface by marking it as connected."""
    # First get interface from device
    interface = self.get_interface_by_name_and_device(iface_name, device_name)

    # Check if interface is reserved
    if interface.mark_connected:
        msg = f"The interface: {iface_name} on device: {device_name} is already allocated."
        raise WorkflowStateError(msg)

    # Allocate interface by marking it as connected
    interface.mark_connected = True
    interface.save()

    return interface

free_interface(device_name, iface_name)

Free interface by marking disconnect and disable it.

Source code in gso/services/netbox_client.py 
272
273
274
275
276
277
278
279
280
281
def free_interface(self, device_name: str, iface_name: str) -> Interfaces:
    """Free interface by marking disconnect and disable it."""
    # First get interface from device
    interface = self.get_interface_by_name_and_device(iface_name, device_name)
    interface.mark_connected = False
    interface.enabled = False
    interface.description = ""
    interface.save()

    return interface

detach_interfaces_from_lag(device_name, lag_name)

Detach all interfaces from a LAG.

Source code in gso/services/netbox_client.py 
283
284
285
286
287
288
289
290
291
292
293
294
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()

get_available_lags_in_range(router_id, lag_range)

Return all available LAGs within a given range not assigned to a device.

Source code in gso/services/netbox_client.py 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
def get_available_lags_in_range(self, router_id: UUID, lag_range: range) -> list[str]:
    """Return all available LAGs within a given range 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 in the specified range
    all_feasible_lags = [f"lag-{i}" for i in 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]

get_available_lags(router_id)

Return all available LAG not assigned to a device.

Source code in gso/services/netbox_client.py 
312
313
314
def get_available_lags(self, router_id: UUID) -> list[str]:
    """Return all available LAG not assigned to a device."""
    return self.get_available_lags_in_range(router_id, FEASIBLE_IP_TRUNK_LAG_RANGE)

get_available_services_lags(router_id)

Return all available Edge port LAGs not assigned to a device.

Source code in gso/services/netbox_client.py 
316
317
318
def get_available_services_lags(self, router_id: UUID) -> list[str]:
    """Return all available Edge port LAGs not assigned to a device."""
    return self.get_available_lags_in_range(router_id, FEASIBLE_SERVICES_LAG_RANGE)

calculate_speed_bits_per_sec(speed) staticmethod

Calculate the numeric part from the speed.

Source code in gso/services/netbox_client.py 
320
321
322
323
324
325
@staticmethod
def calculate_speed_bits_per_sec(speed: str) -> int:
    """Calculate the numeric part from the speed."""
    numeric_part = int("".join(filter(str.isdigit, speed)))
    # Convert to bits per second
    return numeric_part * 1000000

get_available_interfaces(router_id, speed)

Return all available interfaces of a device filtered by speed.

Source code in gso/services/netbox_client.py 
327
328
329
330
331
332
333
334
335
336
337
def get_available_interfaces(self, router_id: UUID | UUIDstr, 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,
    )