feat: Customizations, time zone, consoleserverports

Author: snaselj

docs/dev/generator.md

@@ -48,7 +48,7 @@ This is achieved through `adapter.configure_model(content_type: ContentTypeStr)`
 - `extend_content_type`: Define this when a source model extends another source model to merge into a single Nautobot model.
 - `forward_references`: Define to forward references to another content type. This is useful when the source data references a content type that is not directly related to the Nautobot content type. For example, a source data references a `dcim.location` content type, but the Nautobot content type is `dcim.locationtype`.
 - `disable_related_reference`: Define, to disable storing references to this content type to other content types. This is useful for e.g. `ObjectChange` model.
-- `pre_import`: Define a callable to be executed before importing the source data. Can be used to alter or cache the source data before importing. `pre_import(source: RecordData, importer_pass: ImporterPass) -> PreImportResult` is called twice for each source record: on first and second input data iteration.
+- `pre_import_record`: Define a callable to be executed before importing the source data. Can be used to alter or cache the source data before importing. `pre_import_record(source: RecordData, importer_pass: ImporterPass) -> PreImportResult` is called twice for each source record: on first and second input data iteration.
 - `fields`: Define the source fields and how they should be imported. This argument is a dictionary mapping `FieldName` to `SourceFieldDefinition` instances.
     - `SourceFieldDefinition` can be one of:
         - `None`: to ignore the field.

nautobot_netbox_importer/diffsync/adapters/netbox.py

@@ -2,7 +2,7 @@
 
 from gzip import GzipFile
 from pathlib import Path
-from typing import Callable, Generator, NamedTuple, Union
+from typing import Callable, Generator, NamedTuple, Sequence, Union
 from urllib.parse import ParseResult, urlparse
 
 import ijson
@@ -15,6 +15,8 @@
 from nautobot_netbox_importer.generator import SourceAdapter, SourceDataGenerator, SourceRecord
 from nautobot_netbox_importer.summary import Pathable
 
+# from nautobot_netbox_importer.diffsync.models.cables import fix_cables
+
 for _name in (
     "base",
     "circuits",
@@ -52,6 +54,7 @@ class NetBoxImporterOptions(NamedTuple):
     save_json_summary_path: str = ""
     save_text_summary_path: str = ""
     trace_issues: bool = False
+    customizations: Sequence[str] = []
 
 
 AdapterSetupFunction = Callable[[SourceAdapter], None]
@@ -75,6 +78,10 @@ def __init__(self, input_ref: _FileRef, options: NetBoxImporterOptions, job=None
 
         self.options = options
 
+        for name in options.customizations:
+            if name:
+                register_generator_setup(name)
+
         for name in GENERATOR_SETUP_MODULES:
             setup = __import__(name, fromlist=["setup"]).setup
             setup(self)
@@ -86,13 +93,16 @@ def load(self) -> None:
             fix_power_feed_locations(self)
         if self.options.unrack_zero_uheight_devices:
             unrack_zero_uheight_devices(self)
+
+        # fix_cables(self)
+
         self.post_import()
 
     def import_to_nautobot(self) -> None:
         """Import a NetBox export file into Nautobot."""
         commited = False
         try:
-            self._atomic_import()
+            self._atomic_import()  # type: ignore
             commited = True
         except _DryRunException:
             logger.warning("Dry-run mode, no data has been imported.")

nautobot_netbox_importer/diffsync/models/base.py

@@ -74,10 +74,12 @@ def content_types_mapper_importer(source: RecordData, target: DiffSyncBaseModel)
 
 def setup(adapter: SourceAdapter) -> None:
     """Map NetBox base models to Nautobot."""
-    adapter.disable_model("sessions.session", "Nautobot has own sessions, sessions should never cross apps.")
     adapter.disable_model("admin.logentry", "Not directly used in Nautobot.")
-    adapter.disable_model("users.userconfig", "May not have a 1 to 1 translation to Nautobot.")
     adapter.disable_model("auth.permission", "Handled via a Nautobot model and may not be a 1 to 1.")
+    adapter.disable_model("dcim.cablepath", "Recreated in Nautobot on signal when circuit termination is created")
+    adapter.disable_model("extras.imageattachment", "Images are not imported yet.")
+    adapter.disable_model("sessions.session", "Nautobot has own sessions, sessions should never cross apps.")
+    adapter.disable_model("users.userconfig", "May not have a 1 to 1 translation to Nautobot.")
 
     _setup_content_types(adapter)
 

nautobot_netbox_importer/diffsync/models/cables.py

@@ -0,0 +1,190 @@
+# COMPATIBLE_TERMINATION_TYPES = {
+#     "circuittermination": ["interface", "frontport", "rearport", "circuittermination"],
+#     "consoleport": ["consoleserverport", "frontport", "rearport"],
+#     "consoleserverport": ["consoleport", "frontport", "rearport"],
+#     "interface": ["interface", "circuittermination", "frontport", "rearport"],
+#     "frontport": [
+#         "consoleport",
+#         "consoleserverport",
+#         "interface",
+#         "frontport",
+#         "rearport",
+#         "circuittermination",
+#     ],
+#     "powerfeed": ["powerport"],
+#     "poweroutlet": ["powerport"],
+#     "powerport": ["poweroutlet", "powerfeed"],
+#     "rearport": [
+#         "consoleport",
+#         "consoleserverport",
+#         "interface",
+#         "frontport",
+#         "rearport",
+#         "circuittermination",
+#     ],
+# }
+"""DCIM data related functions."""
+
+from nautobot.dcim.constants import COMPATIBLE_TERMINATION_TYPES
+
+from nautobot_netbox_importer.base import ContentTypeStr, Uid
+from nautobot_netbox_importer.generator.source import DiffSyncModel, SourceAdapter, SourceModelWrapper
+
+# TBD: Move to base.py
+DUMMY_UID = "dummy"
+
+_FALLBACK_TERMINATION_TYPE = "circuittermination"
+_CIRCUIT_MODELS = {"circuittermination"}
+
+
+def _get_first_compatible_termination_type(stripped_type: str) -> ContentTypeStr:
+    """Determine the first compatible termination type for a given termination.
+
+    This function identifies the first compatible termination type based on the
+    given termination string, falling back to '_FALLBACK_TERMINATION_TYPE' if
+    no compatibility is found.
+
+    Args:
+        stripped_type (str): The termination type with 'dcim.' prefix removed
+
+    Returns:
+        str: The compatible termination type with 'dcim.' prefix
+
+    Examples:
+        >>> _get_first_compatible_termination_type("interface")
+        'dcim.interface'
+
+        >>> _get_first_compatible_termination_type("poweroutlet")
+        'dcim.powerport'
+
+        >>> _get_first_compatible_termination_type("unknown")
+        'circuits.circuittermination'
+    """
+
+    def get_type(model_name: str) -> ContentTypeStr:
+        return f"circuits.{model_name}" if model_name in _CIRCUIT_MODELS else f"dcim.{model_name}"
+
+    if stripped_type not in COMPATIBLE_TERMINATION_TYPES:
+        return get_type(_FALLBACK_TERMINATION_TYPE)
+
+    types = COMPATIBLE_TERMINATION_TYPES[stripped_type]
+    if _FALLBACK_TERMINATION_TYPE in types:
+        return get_type(_FALLBACK_TERMINATION_TYPE)
+
+    return get_type(types[0])
+
+
+def _get_termination(uid: Uid, type_: ContentTypeStr, other_type: ContentTypeStr) -> tuple[Uid, ContentTypeStr]:
+    """Determine the appropriate termination for a cable side.
+
+    This function evaluates cable termination data and returns correct termination information
+    based on compatibility rules.
+
+    Args:
+        uid (Uid): UID of the current termination
+        type_ (ContentTypeStr): Type of the current termination
+        other_type (ContentTypeStr): Type of the opposite termination
+
+    Returns:
+        tuple[str, str]: A tuple containing (termination_id, termination_type)
+
+    Examples:
+        >>> _get_termination("123", "dcim.interface", "dcim.frontport")
+        ('123', 'dcim.interface')
+
+        >>> _get_termination("", "dcim.interface", "dcim.poweroutlet")
+        ('dummy', 'circuit.circuittermination')
+
+        >>> _get_termination("123", "", "dcim.frontport")
+        ('123', 'dcim.interface')
+
+        >>> _get_termination("123", "dcim.interface", "")
+        ('123', 'dcim.interface')
+
+        >>> _get_termination("", "", "")
+        ('dummy', 'circuit.circuittermination')
+
+        >>> _get_termination("456", "dcim.powerport", "dcim.poweroutlet")
+        ('456', 'dcim.powerport')
+    """
+    type_stripped = type_.split(".")[1] if type_ else ""
+    other_stripped = other_type.split(".")[1] if other_type else ""
+    first_compatible = _get_first_compatible_termination_type(other_stripped)
+
+    if not type_:
+        uid = DUMMY_UID
+        type_ = first_compatible
+
+    if not uid:
+        uid = DUMMY_UID
+
+    if not other_type:
+        return uid, type_
+
+    if type_stripped in COMPATIBLE_TERMINATION_TYPES and other_stripped in COMPATIBLE_TERMINATION_TYPES.get(
+        type_stripped, []
+    ):
+        return uid, type_
+
+    return DUMMY_UID, first_compatible
+
+
+def _update_cable_termination(wrapper: SourceModelWrapper, cable: DiffSyncModel, side: str) -> None:
+    """Update cable termination information for a specific side.
+
+    This function retrieves termination data for the specified side of a cable, determines
+    the appropriate termination using _get_termination(), and updates the cable if needed.
+
+    Args:
+        wrapper (SourceModelWrapper): Model wrapper containing field definitions
+        cable (DiffSyncModel): The cable model to update
+        side (str): Which side of the cable to update ('a' or 'b')
+    """
+    uid = getattr(cable, f"termination_{side}_id", "")
+    type_ = getattr(cable, f"termination_{side}_type", "")
+    other_type = getattr(cable, f"termination_{'b' if side == 'a' else 'a'}_type", "")
+
+    result_uid, result_type = _get_termination(uid, type_, other_type)
+
+    if result_uid == uid and result_type == type_:
+        return
+
+    adapter = wrapper.adapter
+
+    field = wrapper.fields[f"termination_{side}_type"]
+    field.set_nautobot_value(cable, adapter.get_nautobot_content_type_uid(result_type))
+
+    field.add_issue(
+        "UpdatedCableTermination",
+        f"Cable termination {side.upper()} updated from {uid}, {type_} to {result_uid}, {result_type}",
+        cable,
+    )
+
+    if result_uid != DUMMY_UID:
+        return
+
+    type_wrapper = adapter.get_or_create_wrapper(result_type)
+    pk = type_wrapper.get_pk_from_uid(result_uid)
+    wrapper.add_reference(type_wrapper, pk)
+
+    field = wrapper.fields[f"termination_{side}_id"]
+    field.set_nautobot_value(cable, pk)
+
+    # TBD: Cached dummy records needs to be improved to contain required values
+    type_wrapper.cache_record({"id": DUMMY_UID})
+
+
+def fix_cables(adapter: SourceAdapter) -> None:
+    """Fix cables by ensuring proper terminations.
+
+    This function processes all cables from the source adapter and validates/fixes
+    termination information for both sides of each cable.
+
+    Args:
+        adapter (SourceAdapter): The source adapter containing cable data
+    """
+    wrapper = adapter.get_or_create_wrapper("dcim.cable")
+
+    for cable in adapter.get_all(wrapper.nautobot.diffsync_class):
+        _update_cable_termination(wrapper, cable, "a")
+        _update_cable_termination(wrapper, cable, "b")

nautobot_netbox_importer/diffsync/models/custom_fields.py

@@ -8,7 +8,7 @@
     EMPTY_VALUES,
     DiffSyncBaseModel,
     ImporterPass,
-    PreImportResult,
+    PreImportRecordResult,
     SourceAdapter,
     SourceField,
     fields,
@@ -95,14 +95,14 @@ def setup(adapter: SourceAdapter) -> None:
     """Map NetBox custom fields to Nautobot."""
     choice_sets = {}
 
-    def create_choice_set(source: RecordData, importer_pass: ImporterPass) -> PreImportResult:
+    def create_choice_set(source: RecordData, importer_pass: ImporterPass) -> PreImportRecordResult:
         if importer_pass == ImporterPass.DEFINE_STRUCTURE:
             choice_sets[source.get("id")] = [
                 *_convert_choices(source.get("base_choices")),
                 *_convert_choices(source.get("extra_choices")),
             ]
 
-        return PreImportResult.USE_RECORD
+        return PreImportRecordResult.USE_RECORD
 
     def define_choice_set(field: SourceField) -> None:
         def choices_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
@@ -144,7 +144,7 @@ def create_choices(choices: list, custom_field_uid: Uid) -> None:
     # Defined in NetBox but not in Nautobot
     adapter.configure_model(
         "extras.CustomFieldChoiceSet",
-        pre_import=create_choice_set,
+        pre_import_record=create_choice_set,
     )
 
     adapter.configure_model(

nautobot_netbox_importer/diffsync/models/dcim.py

@@ -4,7 +4,13 @@
 from uuid import UUID
 
 from nautobot_netbox_importer.base import RecordData
-from nautobot_netbox_importer.generator import DiffSyncBaseModel, PreImportResult, SourceAdapter, SourceField, fields
+from nautobot_netbox_importer.generator import (
+    DiffSyncBaseModel,
+    PreImportRecordResult,
+    SourceAdapter,
+    SourceField,
+    fields,
+)
 
 from .locations import define_location
 
@@ -39,13 +45,13 @@ def units_importer(source: RecordData, target: DiffSyncBaseModel) -> None:
     field.set_importer(units_importer)
 
 
-def _pre_import_cable_termination(source: RecordData, _) -> PreImportResult:
+def _pre_import_cable_termination(source: RecordData, _) -> PreImportRecordResult:
     cable_end = source.pop("cable_end").lower()
     source["id"] = source.pop("cable")
     source[f"termination_{cable_end}_type"] = source.pop("termination_type")
     source[f"termination_{cable_end}_id"] = source.pop("termination_id")
 
-    return PreImportResult.USE_RECORD
+    return PreImportRecordResult.USE_RECORD
 
 
 def setup(adapter: SourceAdapter) -> None:
@@ -65,11 +71,19 @@ def setup(adapter: SourceAdapter) -> None:
             "role": fields.role(adapter, "dcim.rackrole"),
         },
     )
-    adapter.configure_model("dcim.cable")
+    adapter.configure_model(
+        "dcim.cable",
+        fields={
+            "termination_a_type": "termination_a_type",
+            "termination_a_id": "termination_a_id",
+            "termination_b_type": "termination_b_type",
+            "termination_b_id": "termination_b_id",
+        },
+    )
     adapter.configure_model(
         "dcim.cabletermination",
         extend_content_type="dcim.cable",
-        pre_import=_pre_import_cable_termination,
+        pre_import_record=_pre_import_cable_termination,
     )
     adapter.configure_model(
         "dcim.interface",