fix: minimal fixes for NetBox 4.3 initial support (#116)

Author: ltucker

netbox_diode_plugin/api/compat.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python
+# Copyright 2025 NetBox Labs Inc
+"""Diode NetBox Plugin - API - Compatibility Transformations."""
+
+import logging
+import re
+from collections import defaultdict
+from functools import cache
+
+from django.conf import settings
+from packaging import version
+from utilities.release import load_release_data
+
+logger = logging.getLogger(__name__)
+
+_MIGRATIONS_BY_OBJECT_TYPE = defaultdict(list)
+
+def apply_entity_migrations(data: dict, object_type: str):
+    """
+    Applies migrations to diode entity data prior to diffing to improve compatibility with current NetBox version.
+
+    These represent cases like deprecated fields that have been replaced with new fields, but
+    are supported for backwards compatibility.
+    """
+    for migration in _MIGRATIONS_BY_OBJECT_TYPE.get(object_type, []):
+        logger.debug(f"Applying migration {migration.__name__} for {object_type}")
+        migration(data)
+
+def _register_migration(func, min_version, max_version, object_type):
+    """Registers a migration function."""
+    min_version = version.parse(min_version)
+    max_version = version.parse(max_version) if max_version else None
+    current_version = _current_netbox_version()
+
+    if current_version < min_version:
+        logger.debug(f"Skipping migration {func.__name__} for {object_type}: min version {min_version}")
+        return
+    if max_version and current_version > max_version:
+        logger.debug(f"Skipping migration {func.__name__} for {object_type}: max version {max_version}")
+        return
+
+    logger.debug(f"Registering migration {func.__name__} for {object_type}.")
+    _MIGRATIONS_BY_OBJECT_TYPE[object_type].append(func)
+
+@cache
+def _current_netbox_version():
+    """Returns the current version of NetBox."""
+    try:
+        return version.parse(settings.RELEASE.version)
+    except Exception:
+        logger.exception("Failed to determine current version of NetBox.")
+        return (0, 0, 0)
+
+def diode_migration(min_version: str, max_version: str | None, object_type: str):
+    """Decorator to mark a function as a diode migration."""
+    def decorator(func):
+        _register_migration(func, min_version, max_version, object_type)
+        return func
+    return decorator
+
+@diode_migration(min_version="4.3.0", max_version=None, object_type="ipam.service")
+def _migrate_service_parent_object(data: dict):
+    """Transforms ipam.service device and virtual_machine references to parent_object."""
+    device = data.pop("device", None)
+    if device:
+        if data.get("parent_object_device") is None:
+            data["parent_object_device"] = device
+        # else ignored.
+
+    virtual_machine = data.pop("virtual_machine", None)
+    if virtual_machine:
+        if data.get("parent_object_virtual_machine") is None:
+            data["parent_object_virtual_machine"] = virtual_machine
+        # else ignored.
+
+@diode_migration(min_version="4.3.0", max_version=None, object_type="tenancy.contact")
+def _migrate_contact_group(data: dict):
+    """Transforms tenancy.contact group references to groups."""
+    group = data.pop("group", None)
+    if group:
+        if data.get("groups") is None:
+            data["groups"] = [group]
+        # else ignored.

netbox_diode_plugin/api/plugin_utils.py

@@ -544,6 +544,8 @@ class RefInfo:
     'ipam.service': {
         'device': RefInfo(object_type='dcim.device', field_name='device'),
         'ipaddresses': RefInfo(object_type='ipam.ipaddress', field_name='ipaddresses', is_many=True),
+        'parent_object_device': RefInfo(object_type='dcim.device', field_name='parent_object', is_generic=True),
+        'parent_object_virtual_machine': RefInfo(object_type='virtualization.virtualmachine', field_name='parent_object', is_generic=True),
         'tags': RefInfo(object_type='extras.tag', field_name='tags', is_many=True),
         'virtual_machine': RefInfo(object_type='virtualization.virtualmachine', field_name='virtual_machine'),
     },
@@ -576,6 +578,7 @@ class RefInfo:
     },
     'tenancy.contact': {
         'group': RefInfo(object_type='tenancy.contactgroup', field_name='group'),
+        'groups': RefInfo(object_type='tenancy.contactgroup', field_name='groups', is_many=True),
         'tags': RefInfo(object_type='extras.tag', field_name='tags', is_many=True),
     },
     'tenancy.contactassignment': {
@@ -948,13 +951,13 @@ def get_json_ref_info(object_type: str|Type[models.Model], json_field_name: str)
     'ipam.rir': frozenset(['custom_fields', 'description', 'is_private', 'name', 'slug', 'tags']),
     'ipam.role': frozenset(['custom_fields', 'description', 'name', 'slug', 'tags', 'weight']),
     'ipam.routetarget': frozenset(['comments', 'custom_fields', 'description', 'name', 'tags', 'tenant']),
-    'ipam.service': frozenset(['comments', 'custom_fields', 'description', 'device', 'ipaddresses', 'name', 'ports', 'protocol', 'tags', 'virtual_machine']),
+    'ipam.service': frozenset(['comments', 'custom_fields', 'description', 'device', 'ipaddresses', 'name', 'parent_object_id', 'parent_object_type', 'ports', 'protocol', 'tags', 'virtual_machine']),
     'ipam.vlan': frozenset(['comments', 'custom_fields', 'description', 'group', 'name', 'qinq_role', 'qinq_svlan', 'role', 'site', 'status', 'tags', 'tenant', 'vid']),
     'ipam.vlangroup': frozenset(['custom_fields', 'description', 'name', 'scope_id', 'scope_type', 'slug', 'tags', 'vid_ranges']),
     'ipam.vlantranslationpolicy': frozenset(['description', 'name']),
     'ipam.vlantranslationrule': frozenset(['description', 'local_vid', 'policy', 'remote_vid']),
     'ipam.vrf': frozenset(['comments', 'custom_fields', 'description', 'enforce_unique', 'export_targets', 'import_targets', 'name', 'rd', 'tags', 'tenant']),
-    'tenancy.contact': frozenset(['address', 'comments', 'custom_fields', 'description', 'email', 'group', 'link', 'name', 'phone', 'tags', 'title']),
+    'tenancy.contact': frozenset(['address', 'comments', 'custom_fields', 'description', 'email', 'group', 'groups', 'link', 'name', 'phone', 'tags', 'title']),
     'tenancy.contactassignment': frozenset(['contact', 'custom_fields', 'object_id', 'object_type', 'priority', 'role', 'tags']),
     'tenancy.contactgroup': frozenset(['custom_fields', 'description', 'name', 'parent', 'slug', 'tags']),
     'tenancy.contactrole': frozenset(['custom_fields', 'description', 'name', 'slug', 'tags']),

netbox_diode_plugin/api/transformer.py

@@ -17,6 +17,7 @@
 from rest_framework import serializers
 
 from .common import NON_FIELD_ERRORS, AutoSlug, ChangeSetException, UnresolvedReference, harmonize_formats, sort_ints_first
+from .compat import apply_entity_migrations
 from .matcher import find_existing_object, fingerprints
 from .plugin_utils import (
     CUSTOM_FIELD_OBJECT_REFERENCE_TYPE,
@@ -125,6 +126,7 @@ def _transform_proto_json_1(proto_json: dict, object_type: str, context=None) ->
     # handle camelCase protoJSON if provided...
     proto_json = _ensure_snake_case(proto_json, object_type)
     apply_format_transformations(proto_json, object_type)
+    apply_entity_migrations(proto_json, object_type)
 
     # context pushed down from parent nodes
     if context is not None: