MrBlake/Netbox-Utilities
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: remove existing patchpanel backfill

Browse Source
This commit is contained in:
MrBlake
2026-08-06 16:03:58 +02:00
parent 8c6c48531e
commit 66d13292f2
11 changed files with 30 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files
netbox_utilities/migrations/0004_patchpanel_port_mappings.py
+3 -46
View File
@@ -1,55 +1,12 @@
import re
from django.db import migrations
def natural_port_key(port):
parts = tuple(
int(part) if part.isdigit() else part.casefold()
for part in re.split(r"(\d+)", str(port.name))
)
return parts, port.pk
def synchronize_existing_patchpanels(apps, schema_editor):
Device = apps.get_model("dcim", "Device")
FrontPort = apps.get_model("dcim", "FrontPort")
PortMapping = apps.get_model("dcim", "PortMapping")
RearPort = apps.get_model("dcim", "RearPort")
database = schema_editor.connection.alias
device_ids = Device.objects.using(database).filter(role__name__iexact="Patchpanel").values_list("pk", flat=True)
for device_id in device_ids.iterator():
front_ports = sorted(
FrontPort.objects.using(database).filter(device_id=device_id).only("pk", "name"),
key=natural_port_key,
)
rear_ports = sorted(
RearPort.objects.using(database).filter(device_id=device_id).only("pk", "name"),
key=natural_port_key,
)
PortMapping.objects.using(database).filter(device_id=device_id).delete()
PortMapping.objects.using(database).bulk_create(
[
PortMapping(
device_id=device_id,
front_port_id=front_port.pk,
rear_port_id=rear_port.pk,
front_port_position=1,
rear_port_position=1,
)
for front_port, rear_port in zip(front_ports, rear_ports)
],
batch_size=1000,
)
class Migration(migrations.Migration):
dependencies = [
("dcim", "0237_module_remove_local_context_data"),
("netbox_utilities", "0003_navigationpreference_sidebar_layout"),
]
operations = [
migrations.RunPython(synchronize_existing_patchpanels, migrations.RunPython.noop),
]
# Kept in the migration graph for installations which already recorded it.
# Existing devices must not be modified automatically.
operations = []
netbox_utilities/migrations/0005_fix_patchpanel_port_number_mappings.py
+3 -73
View File
@@ -1,81 +1,11 @@
import re
from collections import defaultdict
from django.db import migrations
def natural_port_key(port):
parts = tuple(
int(part) if part.isdigit() else part.casefold()
for part in re.split(r"(\d+)", str(port.name))
)
return parts, port.pk
def port_identifier(port):
name = str(port.name).casefold()
name = re.sub(r"\b(?:front|rear)(?:[\s_-]*port)?(?=\b|\d)", " ", name)
name = re.sub(r"^(?:fp|rp|f|r)(?=\s*[-_.:/]?\s*\d)", "", name)
tokens = re.findall(r"\d+|[^\W\d_]+", name)
identifier = tuple(int(token) if token.isdigit() else token for token in tokens)
return identifier or None
def pair_ports(front_ports, rear_ports):
front_by_identifier = defaultdict(list)
rear_by_identifier = defaultdict(list)
for port in front_ports:
if identifier := port_identifier(port):
front_by_identifier[identifier].append(port)
for port in rear_ports:
if identifier := port_identifier(port):
rear_by_identifier[identifier].append(port)
pairs = []
for identifier, matching_front_ports in front_by_identifier.items():
pairs.extend(
zip(
sorted(matching_front_ports, key=natural_port_key),
sorted(rear_by_identifier.get(identifier, []), key=natural_port_key),
)
)
return pairs
def fix_existing_patchpanel_mappings(apps, schema_editor):
Device = apps.get_model("dcim", "Device")
FrontPort = apps.get_model("dcim", "FrontPort")
PortMapping = apps.get_model("dcim", "PortMapping")
RearPort = apps.get_model("dcim", "RearPort")
database = schema_editor.connection.alias
device_ids = Device.objects.using(database).filter(role__name__iexact="Patchpanel").values_list("pk", flat=True)
for device_id in device_ids.iterator():
front_ports = FrontPort.objects.using(database).filter(device_id=device_id).only("pk", "name")
rear_ports = RearPort.objects.using(database).filter(device_id=device_id).only("pk", "name")
pairs = pair_ports(front_ports, rear_ports)
PortMapping.objects.using(database).filter(device_id=device_id).delete()
PortMapping.objects.using(database).bulk_create(
[
PortMapping(
device_id=device_id,
front_port_id=front_port.pk,
rear_port_id=rear_port.pk,
front_port_position=1,
rear_port_position=1,
)
for front_port, rear_port in pairs
],
batch_size=1000,
)
class Migration(migrations.Migration):
dependencies = [
("netbox_utilities", "0004_patchpanel_port_mappings"),
]
operations = [
migrations.RunPython(fix_existing_patchpanel_mappings, migrations.RunPython.noop),
]
# Kept in the migration graph for installations which already recorded it.
# Existing devices must not be modified automatically.
operations = []
netbox_utilities/migrations/0006_retrace_existing_patchpanel_paths.py
+3 -11
View File
@@ -1,19 +1,11 @@
from django.db import migrations
def repair_mappings_and_paths(apps, schema_editor):
# Runtime models are intentional here: NetBox's CablePath.retrace() is
# required to refresh the denormalized path data after migration 0005.
from netbox_utilities.patchpanel import repair_existing_patchpanels
repair_existing_patchpanels()
class Migration(migrations.Migration):
dependencies = [
("netbox_utilities", "0005_fix_patchpanel_port_number_mappings"),
]
operations = [
migrations.RunPython(repair_mappings_and_paths, migrations.RunPython.noop),
]
# Kept in the migration graph for installations which already recorded it.
# Existing devices and cable paths must not be modified automatically.
operations = []