Converted all initializers to plugin format

Author: tobiasge

src/netbox_initializers/initializers/__init__.py

@@ -27,7 +27,7 @@
     "device_roles",
     "device_types",
     "devices",
-    "dcim_interfaces",
+    "interfaces",
     "platforms",
     "route_targets",
     "vrfs",
@@ -69,15 +69,18 @@ def load_data(self):
         # Must be implemented by specific subclass
         pass
 
-    def load_yaml(self):
-        yf = Path(f"{self.data_file_path}/{self.data_file_name}")
+    def load_yaml(self, data_file_name=None):
+        if data_file_name:
+            yf = Path(f"{self.data_file_path}/{data_file_name}")
+        else:
+            yf = Path(f"{self.data_file_path}/{self.data_file_name}")
         if not yf.is_file():
             return None
         with yf.open("r") as stream:
             yaml = YAML(typ="safe")
             return yaml.load(stream)
 
-    def pop_custom_fields(params):
+    def pop_custom_fields(self, params):
         if "custom_field_data" in params:
             return params.pop("custom_field_data")
         elif "custom_fields" in params:
@@ -86,7 +89,7 @@ def pop_custom_fields(params):
 
         return None
 
-    def set_custom_fields_values(entity, custom_field_data):
+    def set_custom_fields_values(self, entity, custom_field_data):
         if not custom_field_data:
             return
 
@@ -131,14 +134,58 @@ def split_params(self, params: dict, unique_params: list = None) -> Tuple[dict,
         return matching_params, params
 
 
+class InitializationError(Exception):
+    pass
+
+
 def register_initializer(name: str, initializer: BaseInitializer):
     INITIALIZER_REGISTRY[name] = initializer
 
 
 # All initializers must be imported here, to be registered
+from .aggregates import AggregateInitializer
 from .asns import ASNInitializer
+from .cables import CableInitializer
+from .circuit_types import CircuitTypeInitializer
+from .circuits import CircuitInitializer
+from .cluster_groups import ClusterGroupInitializer
 from .cluster_types import ClusterTypesInitializer
+from .clusters import ClusterInitializer
+from .contact_groups import ContactGroupInitializer
+from .contact_roles import ContactRoleInitializer
+from .contacts import ContactInitializer
+from .custom_fields import CustomFieldInitializer
+from .custom_links import CustomLinkInitializer
+from .device_roles import DeviceRoleInitializer
+from .device_types import DeviceTypeInitializer
+from .devices import DeviceInitializer
 from .groups import GroupInitializer
+from .interfaces import InterfaceInitializer
+from .ip_addresses import IPAddressInitializer
+from .locations import LocationInitializer
+from .manufacturers import ManufacturerInitializer
+from .object_permissions import ObjectPermissionInitializer
+from .platforms import PlatformInitializer
+from .power_feeds import PowerFeedInitializer
+from .power_panels import PowerPanelInitializer
+from .prefix_vlan_roles import RoleInitializer
+from .prefixes import PrefixInitializer
+from .primary_ips import PrimaryIPInitializer
+from .providers import ProviderInitializer
+from .rack_roles import RackRoleInitializer
+from .racks import RackInitializer
+from .regions import RegionInitializer
 from .rirs import RIRInitializer
+from .route_targets import RouteTargetInitializer
 from .services import ServiceInitializer
+from .sites import SiteInitializer
+from .tags import TagInitializer
+from .tenant_groups import TenantGroupInitializer
+from .tenants import TenantInitializer
 from .users import UserInitializer
+from .virtual_machines import VirtualMachineInitializer
+from .virtualization_interfaces import VMInterfaceInitializer
+from .vlan_groups import VLANGroupInitializer
+from .vlans import VLANInitializer
+from .vrfs import VRFInitializer
+from .webhooks import WebhookInitializer

src/netbox_initializers/initializers/aggregates.py

@@ -0,0 +1,50 @@
+from ipam.models import RIR, Aggregate
+from netaddr import IPNetwork
+from tenancy.models import Tenant
+
+from . import BaseInitializer, register_initializer
+
+MATCH_PARAMS = ["prefix", "rir"]
+REQUIRED_ASSOCS = {"rir": (RIR, "name")}
+OPTIONAL_ASSOCS = {
+    "tenant": (Tenant, "name"),
+}
+
+
+class AggregateInitializer(BaseInitializer):
+    data_file_name = "aggregates.yml"
+
+    def load_data(self):
+        aggregates = self.load_yaml()
+        if aggregates is None:
+            return
+        for params in aggregates:
+            custom_field_data = self.pop_custom_fields(params)
+
+            params["prefix"] = IPNetwork(params["prefix"])
+
+            for assoc, details in REQUIRED_ASSOCS.items():
+                model, field = details
+                query = {field: params.pop(assoc)}
+
+                params[assoc] = model.objects.get(**query)
+
+            for assoc, details in OPTIONAL_ASSOCS.items():
+                if assoc in params:
+                    model, field = details
+                    query = {field: params.pop(assoc)}
+
+                    params[assoc] = model.objects.get(**query)
+
+            matching_params, defaults = self.split_params(params, MATCH_PARAMS)
+            aggregate, created = Aggregate.objects.get_or_create(
+                **matching_params, defaults=defaults
+            )
+
+            if created:
+                print("🗞️ Created Aggregate", aggregate.prefix)
+
+            self.set_custom_fields_values(aggregate, custom_field_data)
+
+
+register_initializer("aggregates", AggregateInitializer)

src/netbox_initializers/initializers/cables.py

@@ -0,0 +1,236 @@
+from typing import Tuple
+
+from circuits.models import Circuit, CircuitTermination, ProviderNetwork
+from dcim.models import (
+    Cable,
+    ConsolePort,
+    ConsoleServerPort,
+    FrontPort,
+    Interface,
+    PowerFeed,
+    PowerOutlet,
+    PowerPanel,
+    PowerPort,
+    RearPort,
+    Site,
+)
+from django.contrib.contenttypes.models import ContentType
+from django.db.models import Q
+
+from . import BaseInitializer, register_initializer
+
+CONSOLE_PORT_TERMINATION = ContentType.objects.get_for_model(ConsolePort)
+CONSOLE_SERVER_PORT_TERMINATION = ContentType.objects.get_for_model(ConsoleServerPort)
+FRONT_PORT_TERMINATION = ContentType.objects.get_for_model(FrontPort)
+REAR_PORT_TERMINATION = ContentType.objects.get_for_model(RearPort)
+FRONT_AND_REAR = [FRONT_PORT_TERMINATION, REAR_PORT_TERMINATION]
+POWER_PORT_TERMINATION = ContentType.objects.get_for_model(PowerPort)
+POWER_OUTLET_TERMINATION = ContentType.objects.get_for_model(PowerOutlet)
+POWER_FEED_TERMINATION = ContentType.objects.get_for_model(PowerFeed)
+POWER_TERMINATIONS = [POWER_PORT_TERMINATION, POWER_OUTLET_TERMINATION, POWER_FEED_TERMINATION]
+
+VIRTUAL_INTERFACES = ["bridge", "lag", "virtual"]
+
+
+def get_termination_object(params: dict, side: str):
+    klass = params.pop(f"termination_{side}_class")
+    name = params.pop(f"termination_{side}_name", None)
+    device = params.pop(f"termination_{side}_device", None)
+    feed_params = params.pop(f"termination_{side}_feed", None)
+    circuit_params = params.pop(f"termination_{side}_circuit", {})
+
+    if name and device:
+        termination = klass.objects.get(name=name, device__name=device)
+        return termination
+    elif feed_params:
+        q = {
+            "name": feed_params["power_panel"]["name"],
+            "site__name": feed_params["power_panel"]["site"],
+        }
+        power_panel = PowerPanel.objects.get(**q)
+        termination = PowerFeed.objects.get(name=feed_params["name"], power_panel=power_panel)
+        return termination
+    elif circuit_params:
+        circuit = Circuit.objects.get(cid=circuit_params.pop("cid"))
+        term_side = circuit_params.pop("term_side").upper()
+
+        site_name = circuit_params.pop("site", None)
+        provider_network = circuit_params.pop("provider_network", None)
+
+        if site_name:
+            circuit_params["site"] = Site.objects.get(name=site_name)
+        elif provider_network:
+            circuit_params["provider_network"] = ProviderNetwork.objects.get(name=provider_network)
+        else:
+            raise ValueError(
+                f"⚠️ Missing one of required parameters: 'site' or 'provider_network' "
+                f"for side {term_side} of circuit {circuit}"
+            )
+
+        termination, created = CircuitTermination.objects.get_or_create(
+            circuit=circuit, term_side=term_side, defaults=circuit_params
+        )
+        if created:
+            print(f"⚡ Created new CircuitTermination {termination}")
+
+        return termination
+
+    raise ValueError(
+        f"⚠️ Missing parameters for termination_{side}. "
+        "Need termination_{side}_name AND termination_{side}_device OR termination_{side}_circuit"
+    )
+
+
+def get_termination_class_by_name(port_class: str):
+    if not port_class:
+        return Interface
+
+    return globals()[port_class]
+
+
+def cable_in_cables(term_a: tuple, term_b: tuple) -> bool:
+    """Check if cable exist for given terminations.
+    Each tuple should consist termination object and termination type
+    """
+
+    cable = Cable.objects.filter(
+        Q(
+            termination_a_id=term_a[0].id,
+            termination_a_type=term_a[1],
+            termination_b_id=term_b[0].id,
+            termination_b_type=term_b[1],
+        )
+        | Q(
+            termination_a_id=term_b[0].id,
+            termination_a_type=term_b[1],
+            termination_b_id=term_a[0].id,
+            termination_b_type=term_a[1],
+        )
+    )
+    return cable.exists()
+
+
+def check_termination_types(type_a, type_b) -> Tuple[bool, str]:
+    if type_a in POWER_TERMINATIONS and type_b in POWER_TERMINATIONS:
+        if type_a == type_b:
+            return False, "Can't connect the same power terminations together"
+        elif (
+            type_a == POWER_OUTLET_TERMINATION
+            and type_b == POWER_FEED_TERMINATION
+            or type_a == POWER_FEED_TERMINATION
+            and type_b == POWER_OUTLET_TERMINATION
+        ):
+            return False, "PowerOutlet can't be connected with PowerFeed"
+    elif type_a in POWER_TERMINATIONS or type_b in POWER_TERMINATIONS:
+        return False, "Can't mix power terminations with port terminations"
+    elif type_a in FRONT_AND_REAR or type_b in FRONT_AND_REAR:
+        return True, ""
+    elif (
+        type_a == CONSOLE_PORT_TERMINATION
+        and type_b != CONSOLE_SERVER_PORT_TERMINATION
+        or type_b == CONSOLE_PORT_TERMINATION
+        and type_a != CONSOLE_SERVER_PORT_TERMINATION
+    ):
+        return False, "ConsolePorts can only be connected to ConsoleServerPorts or Front/Rear ports"
+    return True, ""
+
+
+def get_cable_name(termination_a: tuple, termination_b: tuple) -> str:
+    """Returns name of a cable in format:
+    device_a interface_a <---> interface_b device_b
+    or for circuits:
+    circuit_a termination_a <---> termination_b circuit_b
+    """
+    cable_name = []
+
+    for is_side_b, termination in enumerate([termination_a, termination_b]):
+        try:
+            power_panel_id = getattr(termination[0], "power_panel_id", None)
+            if power_panel_id:
+                power_feed = PowerPanel.objects.get(id=power_panel_id)
+                segment = [f"{power_feed}", f"{termination[0]}"]
+            else:
+                segment = [f"{termination[0].device}", f"{termination[0]}"]
+        except AttributeError:
+            segment = [f"{termination[0].circuit.cid}", f"{termination[0]}"]
+
+        if is_side_b:
+            segment.reverse()
+
+        cable_name.append(" ".join(segment))
+
+    return " <---> ".join(cable_name)
+
+
+def check_interface_types(*args):
+    for termination in args:
+        try:
+            if termination.type in VIRTUAL_INTERFACES:
+                raise Exception(
+                    f"⚠️ Virtual interfaces are not supported for cabling. "
+                    f"Termination {termination.device} {termination} {termination.type}"
+                )
+        except AttributeError:
+            # CircuitTermination doesn't have a type field
+            pass
+
+
+def check_terminations_are_free(*args):
+    any_failed = False
+    for termination in args:
+        if termination.cable_id:
+            any_failed = True
+            print(
+                f"⚠️ Termination {termination} is already occupied "
+                f"with cable #{termination.cable_id}"
+            )
+    if any_failed:
+        raise Exception("⚠️ At least one end of the cable is already occupied.")
+
+
+class CableInitializer(BaseInitializer):
+    data_file_name = "cables.yml"
+
+    def load_data(self):
+        cables = self.load_yaml()
+        if cables is None:
+            return
+        for params in cables:
+            params["termination_a_class"] = get_termination_class_by_name(
+                params.get("termination_a_class")
+            )
+            params["termination_b_class"] = get_termination_class_by_name(
+                params.get("termination_b_class")
+            )
+
+            term_a = get_termination_object(params, side="a")
+            term_b = get_termination_object(params, side="b")
+
+            check_interface_types(term_a, term_b)
+
+            term_a_ct = ContentType.objects.get_for_model(term_a)
+            term_b_ct = ContentType.objects.get_for_model(term_b)
+
+            types_ok, msg = check_termination_types(term_a_ct, term_b_ct)
+            cable_name = get_cable_name((term_a, term_a_ct), (term_b, term_b_ct))
+
+            if not types_ok:
+                print(f"⚠️ Invalid termination types for {cable_name}. {msg}")
+                continue
+
+            if cable_in_cables((term_a, term_a_ct), (term_b, term_b_ct)):
+                continue
+
+            check_terminations_are_free(term_a, term_b)
+
+            params["termination_a_id"] = term_a.id
+            params["termination_b_id"] = term_b.id
+            params["termination_a_type"] = term_a_ct
+            params["termination_b_type"] = term_b_ct
+
+            cable = Cable.objects.create(**params)
+
+            print(f"🧷 Created cable {cable} {cable_name}")
+
+
+register_initializer("cables", CableInitializer)