NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-07-26 18:38:38 -06:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

Move Module & ModuleType models to a separate file

Browse Source
This commit is contained in:
Jeremy Stretch 2025-03-25 13:50:30 -04:00
parent 1508e3a770
commit 65ad972a1c
4 changed files with 302 additions and 280 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
netbox/dcim/models/__init__.py
View File

@ -2,6 +2,7 @@ from .cables import *
from .device_component_templates import *
from .device_components import *
from .devices import *
from .modules import *
from .power import *
from .racks import *
from .sites import *

282
netbox/dcim/models/devices.py
View File

@ -19,6 +19,7 @@ from core.models import ObjectType
from dcim.choices import *
from dcim.constants import *
from dcim.fields import MACAddressField
from dcim.utils import update_interface_bridges
from extras.models import ConfigContextModel, CustomField
from extras.querysets import ConfigContextModelQuerySet
from netbox.choices import ColorChoices
@ -30,6 +31,7 @@ from utilities.fields import ColorField, CounterCacheField
from utilities.tracking import TrackingModelMixin
from .device_components import *
from .mixins import RenderConfigMixin
from .modules import Module
__all__ = (
@ -38,8 +40,6 @@ __all__ = (
'DeviceType',
'MACAddress',
'Manufacturer',
'Module',
'ModuleType',
'Platform',
'VirtualChassis',
'VirtualDeviceContext',
@ -367,103 +367,6 @@ class DeviceType(ImageAttachmentsMixin, PrimaryModel, WeightMixin):
return self.subdevice_role == SubdeviceRoleChoices.ROLE_CHILD
class ModuleType(ImageAttachmentsMixin, PrimaryModel, WeightMixin):
"""
A ModuleType represents a hardware element that can be installed within a device and which houses additional
components; for example, a line card within a chassis-based switch such as the Cisco Catalyst 6500. Like a
DeviceType, each ModuleType can have console, power, interface, and pass-through port templates assigned to it. It
cannot, however house device bays or module bays.
"""
manufacturer = models.ForeignKey(
to='dcim.Manufacturer',
on_delete=models.PROTECT,
related_name='module_types'
)
model = models.CharField(
verbose_name=_('model'),
max_length=100
)
part_number = models.CharField(
verbose_name=_('part number'),
max_length=50,
blank=True,
help_text=_('Discrete part number (optional)')
)
airflow = models.CharField(
verbose_name=_('airflow'),
max_length=50,
choices=ModuleAirflowChoices,
blank=True,
null=True
)
clone_fields = ('manufacturer', 'weight', 'weight_unit', 'airflow')
prerequisite_models = (
'dcim.Manufacturer',
)
class Meta:
ordering = ('manufacturer', 'model')
constraints = (
models.UniqueConstraint(
fields=('manufacturer', 'model'),
name='%(app_label)s_%(class)s_unique_manufacturer_model'
),
)
verbose_name = _('module type')
verbose_name_plural = _('module types')
def __str__(self):
return self.model
@property
def full_name(self):
return f"{self.manufacturer} {self.model}"
def to_yaml(self):
data = {
'manufacturer': self.manufacturer.name,
'model': self.model,
'part_number': self.part_number,
'description': self.description,
'weight': float(self.weight) if self.weight is not None else None,
'weight_unit': self.weight_unit,
'comments': self.comments,
}
# Component templates
if self.consoleporttemplates.exists():
data['console-ports'] = [
c.to_yaml() for c in self.consoleporttemplates.all()
]
if self.consoleserverporttemplates.exists():
data['console-server-ports'] = [
c.to_yaml() for c in self.consoleserverporttemplates.all()
]
if self.powerporttemplates.exists():
data['power-ports'] = [
c.to_yaml() for c in self.powerporttemplates.all()
]
if self.poweroutlettemplates.exists():
data['power-outlets'] = [
c.to_yaml() for c in self.poweroutlettemplates.all()
]
if self.interfacetemplates.exists():
data['interfaces'] = [
c.to_yaml() for c in self.interfacetemplates.all()
]
if self.frontporttemplates.exists():
data['front-ports'] = [
c.to_yaml() for c in self.frontporttemplates.all()
]
if self.rearporttemplates.exists():
data['rear-ports'] = [
c.to_yaml() for c in self.rearporttemplates.all()
]
return yaml.dump(dict(data), sort_keys=False)
#
# Devices
#
@ -526,23 +429,6 @@ class Platform(OrganizationalModel):
verbose_name_plural = _('platforms')
def update_interface_bridges(device, interface_templates, module=None):
"""
Used for device and module instantiation. Iterates all InterfaceTemplates with a bridge assigned
and applies it to the actual interfaces.
"""
for interface_template in interface_templates.exclude(bridge=None):
interface = Interface.objects.get(device=device, name=interface_template.resolve_name(module=module))
if interface_template.bridge:
interface.bridge = Interface.objects.get(
device=device,
name=interface_template.bridge.resolve_name(module=module)
)
interface.full_clean()
interface.save()
class Device(
ContactsMixin,
ImageAttachmentsMixin,
@ -1155,170 +1041,6 @@ class Device(
return round(total_weight / 1000, 2)
class Module(PrimaryModel, ConfigContextModel):
"""
A Module represents a field-installable component within a Device which may itself hold multiple device components
(for example, a line card within a chassis switch). Modules are instantiated from ModuleTypes.
"""
device = models.ForeignKey(
to='dcim.Device',
on_delete=models.CASCADE,
related_name='modules'
)
module_bay = models.OneToOneField(
to='dcim.ModuleBay',
on_delete=models.CASCADE,
related_name='installed_module'
)
module_type = models.ForeignKey(
to='dcim.ModuleType',
on_delete=models.PROTECT,
related_name='instances'
)
status = models.CharField(
verbose_name=_('status'),
max_length=50,
choices=ModuleStatusChoices,
default=ModuleStatusChoices.STATUS_ACTIVE
)
serial = models.CharField(
max_length=50,
blank=True,
verbose_name=_('serial number')
)
asset_tag = models.CharField(
max_length=50,
blank=True,
null=True,
unique=True,
verbose_name=_('asset tag'),
help_text=_('A unique tag used to identify this device')
)
clone_fields = ('device', 'module_type', 'status')
class Meta:
ordering = ('module_bay',)
verbose_name = _('module')
verbose_name_plural = _('modules')
def __str__(self):
return f'{self.module_bay.name}: {self.module_type} ({self.pk})'
def get_status_color(self):
return ModuleStatusChoices.colors.get(self.status)
def clean(self):
super().clean()
if hasattr(self, "module_bay") and (self.module_bay.device != self.device):
raise ValidationError(
_("Module must be installed within a module bay belonging to the assigned device ({device}).").format(
device=self.device
)
)
# Check for recursion
module = self
module_bays = []
modules = []
while module:
if module.pk in modules or module.module_bay.pk in module_bays:
raise ValidationError(_("A module bay cannot belong to a module installed within it."))
modules.append(module.pk)
module_bays.append(module.module_bay.pk)
module = module.module_bay.module if module.module_bay else None
def save(self, *args, **kwargs):
is_new = self.pk is None
super().save(*args, **kwargs)
adopt_components = getattr(self, '_adopt_components', False)
disable_replication = getattr(self, '_disable_replication', False)
# We skip adding components if the module is being edited or
# both replication and component adoption is disabled
if not is_new or (disable_replication and not adopt_components):
return
# Iterate all component types
for templates, component_attribute, component_model in [
("consoleporttemplates", "consoleports", ConsolePort),
("consoleserverporttemplates", "consoleserverports", ConsoleServerPort),
("interfacetemplates", "interfaces", Interface),
("powerporttemplates", "powerports", PowerPort),
("poweroutlettemplates", "poweroutlets", PowerOutlet),
("rearporttemplates", "rearports", RearPort),
("frontporttemplates", "frontports", FrontPort),
("modulebaytemplates", "modulebays", ModuleBay),
]:
create_instances = []
update_instances = []
# Prefetch installed components
installed_components = {
component.name: component
for component in getattr(self.device, component_attribute).filter(module__isnull=True)
}
# Get the template for the module type.
for template in getattr(self.module_type, templates).all():
template_instance = template.instantiate(device=self.device, module=self)
if adopt_components:
existing_item = installed_components.get(template_instance.name)
# Check if there's a component with the same name already
if existing_item:
# Assign it to the module
existing_item.module = self
update_instances.append(existing_item)
continue
# Only create new components if replication is enabled
if not disable_replication:
create_instances.append(template_instance)
# Set default values for any applicable custom fields
if cf_defaults := CustomField.objects.get_defaults_for_model(component_model):
for component in create_instances:
component.custom_field_data = cf_defaults
if component_model is not ModuleBay:
component_model.objects.bulk_create(create_instances)
# Emit the post_save signal for each newly created object
for component in create_instances:
post_save.send(
sender=component_model,
instance=component,
created=True,
raw=False,
using='default',
update_fields=None
)
else:
# ModuleBays must be saved individually for MPTT
for instance in create_instances:
instance.save()
update_fields = ['module']
component_model.objects.bulk_update(update_instances, update_fields)
# Emit the post_save signal for each updated object
for component in update_instances:
post_save.send(
sender=component_model,
instance=component,
created=False,
raw=False,
using='default',
update_fields=update_fields
)
# Interface bridges have to be set after interface instantiation
update_interface_bridges(self.device, self.module_type.interfacetemplates, self)
#
# Virtual chassis
#

279
netbox/dcim/models/modules.py Normal file
View File

@ -0,0 +1,279 @@
import yaml
from django.core.exceptions import ValidationError
from django.db import models
from django.db.models.signals import post_save
from django.utils.translation import gettext_lazy as _
from dcim.choices import *
from dcim.utils import update_interface_bridges
from extras.models import ConfigContextModel, CustomField
from netbox.models import PrimaryModel
from netbox.models.features import ImageAttachmentsMixin
from netbox.models.mixins import WeightMixin
from .device_components import *
__all__ = (
'Module',
'ModuleType',
)
class ModuleType(ImageAttachmentsMixin, PrimaryModel, WeightMixin):
"""
A ModuleType represents a hardware element that can be installed within a device and which houses additional
components; for example, a line card within a chassis-based switch such as the Cisco Catalyst 6500. Like a
DeviceType, each ModuleType can have console, power, interface, and pass-through port templates assigned to it. It
cannot, however house device bays or module bays.
"""
manufacturer = models.ForeignKey(
to='dcim.Manufacturer',
on_delete=models.PROTECT,
related_name='module_types'
)
model = models.CharField(
verbose_name=_('model'),
max_length=100
)
part_number = models.CharField(
verbose_name=_('part number'),
max_length=50,
blank=True,
help_text=_('Discrete part number (optional)')
)
airflow = models.CharField(
verbose_name=_('airflow'),
max_length=50,
choices=ModuleAirflowChoices,
blank=True,
null=True
)
clone_fields = ('manufacturer', 'weight', 'weight_unit', 'airflow')
prerequisite_models = (
'dcim.Manufacturer',
)
class Meta:
ordering = ('manufacturer', 'model')
constraints = (
models.UniqueConstraint(
fields=('manufacturer', 'model'),
name='%(app_label)s_%(class)s_unique_manufacturer_model'
),
)
verbose_name = _('module type')
verbose_name_plural = _('module types')
def __str__(self):
return self.model
@property
def full_name(self):
return f"{self.manufacturer} {self.model}"
def to_yaml(self):
data = {
'manufacturer': self.manufacturer.name,
'model': self.model,
'part_number': self.part_number,
'description': self.description,
'weight': float(self.weight) if self.weight is not None else None,
'weight_unit': self.weight_unit,
'comments': self.comments,
}
# Component templates
if self.consoleporttemplates.exists():
data['console-ports'] = [
c.to_yaml() for c in self.consoleporttemplates.all()
]
if self.consoleserverporttemplates.exists():
data['console-server-ports'] = [
c.to_yaml() for c in self.consoleserverporttemplates.all()
]
if self.powerporttemplates.exists():
data['power-ports'] = [
c.to_yaml() for c in self.powerporttemplates.all()
]
if self.poweroutlettemplates.exists():
data['power-outlets'] = [
c.to_yaml() for c in self.poweroutlettemplates.all()
]
if self.interfacetemplates.exists():
data['interfaces'] = [
c.to_yaml() for c in self.interfacetemplates.all()
]
if self.frontporttemplates.exists():
data['front-ports'] = [
c.to_yaml() for c in self.frontporttemplates.all()
]
if self.rearporttemplates.exists():
data['rear-ports'] = [
c.to_yaml() for c in self.rearporttemplates.all()
]
return yaml.dump(dict(data), sort_keys=False)
class Module(PrimaryModel, ConfigContextModel):
"""
A Module represents a field-installable component within a Device which may itself hold multiple device components
(for example, a line card within a chassis switch). Modules are instantiated from ModuleTypes.
"""
device = models.ForeignKey(
to='dcim.Device',
on_delete=models.CASCADE,
related_name='modules'
)
module_bay = models.OneToOneField(
to='dcim.ModuleBay',
on_delete=models.CASCADE,
related_name='installed_module'
)
module_type = models.ForeignKey(
to='dcim.ModuleType',
on_delete=models.PROTECT,
related_name='instances'
)
status = models.CharField(
verbose_name=_('status'),
max_length=50,
choices=ModuleStatusChoices,
default=ModuleStatusChoices.STATUS_ACTIVE
)
serial = models.CharField(
max_length=50,
blank=True,
verbose_name=_('serial number')
)
asset_tag = models.CharField(
max_length=50,
blank=True,
null=True,
unique=True,
verbose_name=_('asset tag'),
help_text=_('A unique tag used to identify this device')
)
clone_fields = ('device', 'module_type', 'status')
class Meta:
ordering = ('module_bay',)
verbose_name = _('module')
verbose_name_plural = _('modules')
def __str__(self):
return f'{self.module_bay.name}: {self.module_type} ({self.pk})'
def get_status_color(self):
return ModuleStatusChoices.colors.get(self.status)
def clean(self):
super().clean()
if hasattr(self, "module_bay") and (self.module_bay.device != self.device):
raise ValidationError(
_("Module must be installed within a module bay belonging to the assigned device ({device}).").format(
device=self.device
)
)
# Check for recursion
module = self
module_bays = []
modules = []
while module:
if module.pk in modules or module.module_bay.pk in module_bays:
raise ValidationError(_("A module bay cannot belong to a module installed within it."))
modules.append(module.pk)
module_bays.append(module.module_bay.pk)
module = module.module_bay.module if module.module_bay else None
def save(self, *args, **kwargs):
is_new = self.pk is None
super().save(*args, **kwargs)
adopt_components = getattr(self, '_adopt_components', False)
disable_replication = getattr(self, '_disable_replication', False)
# We skip adding components if the module is being edited or
# both replication and component adoption is disabled
if not is_new or (disable_replication and not adopt_components):
return
# Iterate all component types
for templates, component_attribute, component_model in [
("consoleporttemplates", "consoleports", ConsolePort),
("consoleserverporttemplates", "consoleserverports", ConsoleServerPort),
("interfacetemplates", "interfaces", Interface),
("powerporttemplates", "powerports", PowerPort),
("poweroutlettemplates", "poweroutlets", PowerOutlet),
("rearporttemplates", "rearports", RearPort),
("frontporttemplates", "frontports", FrontPort),
("modulebaytemplates", "modulebays", ModuleBay),
]:
create_instances = []
update_instances = []
# Prefetch installed components
installed_components = {
component.name: component
for component in getattr(self.device, component_attribute).filter(module__isnull=True)
}
# Get the template for the module type.
for template in getattr(self.module_type, templates).all():
template_instance = template.instantiate(device=self.device, module=self)
if adopt_components:
existing_item = installed_components.get(template_instance.name)
# Check if there's a component with the same name already
if existing_item:
# Assign it to the module
existing_item.module = self
update_instances.append(existing_item)
continue
# Only create new components if replication is enabled
if not disable_replication:
create_instances.append(template_instance)
# Set default values for any applicable custom fields
if cf_defaults := CustomField.objects.get_defaults_for_model(component_model):
for component in create_instances:
component.custom_field_data = cf_defaults
if component_model is not ModuleBay:
component_model.objects.bulk_create(create_instances)
# Emit the post_save signal for each newly created object
for component in create_instances:
post_save.send(
sender=component_model,
instance=component,
created=True,
raw=False,
using='default',
update_fields=None
)
else:
# ModuleBays must be saved individually for MPTT
for instance in create_instances:
instance.save()
update_fields = ['module']
component_model.objects.bulk_update(update_instances, update_fields)
# Emit the post_save signal for each updated object
for component in update_instances:
post_save.send(
sender=component_model,
instance=component,
created=False,
raw=False,
using='default',
update_fields=update_fields
)
# Interface bridges have to be set after interface instantiation
update_interface_bridges(self.device, self.module_type.interfacetemplates, self)

20
netbox/dcim/utils.py
View File

@ -1,3 +1,4 @@
from django.apps import apps
from django.contrib.contenttypes.models import ContentType
from django.db import transaction
@ -56,3 +57,22 @@ def rebuild_paths(terminations):
for cp in cable_paths:
cp.delete()
create_cablepath(cp.origins)
def update_interface_bridges(device, interface_templates, module=None):
"""
Used for device and module instantiation. Iterates all InterfaceTemplates with a bridge assigned
and applies it to the actual interfaces.
"""
Interface = apps.get_model('dcim', 'Interface')
for interface_template in interface_templates.exclude(bridge=None):
interface = Interface.objects.get(device=device, name=interface_template.resolve_name(module=module))
if interface_template.bridge:
interface.bridge = Interface.objects.get(
device=device,
name=interface_template.bridge.resolve_name(module=module)
)
interface.full_clean()
interface.save()