Merge pull request #211 from netboxlabs/195-recursion

195 recursion

Author: bctiemann

netbox_custom_objects/migrations/0003_customobjecttype_object_type.py

@@ -0,0 +1,54 @@
+# Generated by Django 5.2.5 on 2025-09-04 16:34
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+def populate_object_type_field(apps, schema_editor):
+    """
+    Populate the object_type field for existing CustomObjectType instances.
+    """
+    CustomObjectType = apps.get_model('netbox_custom_objects', 'CustomObjectType')
+    ObjectType = apps.get_model('core', 'ObjectType')
+    app_label = CustomObjectType._meta.app_label
+
+    for custom_object_type in CustomObjectType.objects.all():
+        content_type_name = f"Table{custom_object_type.id}Model".lower()
+        try:
+            object_type = ObjectType.objects.get(app_label=app_label, model=content_type_name)
+            custom_object_type.object_type = object_type
+            custom_object_type.save(update_fields=['object_type'])
+        except ObjectType.DoesNotExist:
+            # If ObjectType doesn't exist, create it
+            object_type = ObjectType.objects.create(
+                app_label=app_label,
+                model=content_type_name
+            )
+            custom_object_type.object_type = object_type
+            custom_object_type.save(update_fields=['object_type'])
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("core", "0018_concrete_objecttype"),
+        ("netbox_custom_objects", "0002_customobjecttype_version"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="customobjecttype",
+            name="object_type",
+            field=models.OneToOneField(
+                blank=True,
+                editable=False,
+                null=True,
+                on_delete=django.db.models.deletion.CASCADE,
+                related_name="custom_object_types",
+                to="core.objecttype",
+            ),
+        ),
+        migrations.RunPython(
+            populate_object_type_field,
+        ),
+    ]

netbox_custom_objects/models.py

@@ -14,7 +14,8 @@
 from django.db import connection, IntegrityError, models, transaction
 from django.db.models import Q
 from django.db.models.functions import Lower
-from django.db.models.signals import pre_delete
+from django.db.models.signals import pre_delete, post_save
+from django.dispatch import receiver
 from django.urls import reverse
 from django.utils.translation import gettext_lazy as _
 from core.signals import handle_deleted_object
@@ -187,7 +188,14 @@ class CustomObjectType(PrimaryModel):
     verbose_name = models.CharField(max_length=100, blank=True)
     verbose_name_plural = models.CharField(max_length=100, blank=True)
     slug = models.SlugField(max_length=100, unique=True, db_index=True)
-
+    object_type = models.OneToOneField(
+        ObjectType,
+        on_delete=models.CASCADE,
+        related_name="custom_object_types",
+        null=True,
+        blank=True,
+        editable=False
+    )
     class Meta:
         verbose_name = "Custom Object Type"
         ordering = ("name",)
@@ -302,29 +310,6 @@ def get_list_url(self):
     def get_table_model_name(cls, table_id):
         return f"Table{table_id}Model"
 
-    @property
-    def content_type(self):
-        try:
-            return self.get_or_create_content_type()
-        except Exception:
-            # If we still can't get it, return None
-            return None
-
-    def get_or_create_content_type(self):
-        """
-        Get or create the ObjectType for this CustomObjectType.
-        This ensures the ObjectType is immediately available in the current transaction.
-        """
-        content_type_name = self.get_table_model_name(self.id).lower()
-        try:
-            return ObjectType.objects.get(app_label=APP_LABEL, model=content_type_name)
-        except Exception:
-            # Create the ObjectType and ensure it's immediately available
-            ct = ObjectType.objects.create(app_label=APP_LABEL, model=content_type_name)
-            # Force a refresh to ensure it's available in the current transaction
-            ct.refresh_from_db()
-            return ct
-
     def _fetch_and_generate_field_attrs(
         self,
         fields,
@@ -672,12 +657,11 @@ def create_model(self):
         model = self.get_model()
 
         # Ensure the ContentType exists and is immediately available
-        ct = self.get_or_create_content_type()
         features = get_model_features(model)
-        ct.features = features + ['branching']
-        ct.public = True
-        ct.features = features
-        ct.save()
+        self.object_type.features = features + ['branching']
+        self.object_type.public = True
+        self.object_type.features = features
+        self.object_type.save()
 
         with connection.schema_editor() as schema_editor:
             schema_editor.create_model(model)
@@ -686,7 +670,9 @@ def create_model(self):
 
     def save(self, *args, **kwargs):
         needs_db_create = self._state.adding
+
         super().save(*args, **kwargs)
+
         if needs_db_create:
             self.create_model()
         else:
@@ -700,7 +686,7 @@ def delete(self, *args, **kwargs):
         model = self.get_model()
 
         # Delete all CustomObjectTypeFields that reference this CustomObjectType
-        for field in CustomObjectTypeField.objects.filter(related_object_type=self.content_type):
+        for field in CustomObjectTypeField.objects.filter(related_object_type=self.object_type):
             field.delete()
 
         object_type = ObjectType.objects.get_for_model(model)
@@ -716,6 +702,19 @@ def delete(self, *args, **kwargs):
         pre_delete.connect(handle_deleted_object)
 
 
+@receiver(post_save, sender=CustomObjectType)
+def custom_object_type_post_save_handler(sender, instance, created, **kwargs):
+    if created:
+        # If creating a new object, get or create the ObjectType
+        content_type_name = instance.get_table_model_name(instance.id).lower()
+        ct, created = ObjectType.objects.get_or_create(
+            app_label=APP_LABEL,
+            model=content_type_name
+        )
+        instance.object_type = ct
+        instance.save()
+
+
 class CustomObjectTypeField(CloningMixin, ExportTemplatesMixin, ChangeLoggedModel):
     custom_object_type = models.ForeignKey(
         CustomObjectType, on_delete=models.CASCADE, related_name="fields"
@@ -1124,6 +1123,81 @@ def clean(self):
                     }
                 )
 
+        # Check for recursion in object and multiobject fields
+        if (self.type in (
+            CustomFieldTypeChoices.TYPE_OBJECT,
+            CustomFieldTypeChoices.TYPE_MULTIOBJECT,
+        ) and self.related_object_type_id and
+            self.related_object_type.app_label == APP_LABEL):
+            self._check_recursion()
+
+    def _check_recursion(self):
+        """
+        Check for circular references in object and multiobject fields.
+        Raises ValidationError if recursion is detected.
+        """
+        # Check if this field points to the same custom object type (self-referential)
+        if self.related_object_type_id == self.custom_object_type.object_type_id:
+            return  # Self-referential fields are allowed
+
+        # Get the related custom object type directly from the object_type relationship
+        try:
+            related_custom_object_type = CustomObjectType.objects.get(object_type=self.related_object_type)
+        except CustomObjectType.DoesNotExist:
+            return  # Not a custom object type, no recursion possible
+
+        # Check for circular references by traversing the dependency chain
+        visited = {self.custom_object_type.id}
+        if self._has_circular_reference(related_custom_object_type, visited):
+            raise ValidationError(
+                {
+                    "related_object_type": _(
+                        "Circular reference detected. This field would create a circular dependency "
+                        "between custom object types."
+                    )
+                }
+            )
+
+    def _has_circular_reference(self, custom_object_type, visited):
+        """
+        Recursively check if there's a circular reference by following the dependency chain.
+
+        Args:
+            custom_object_type: The CustomObjectType object to check
+            visited: Set of custom object type IDs already visited in this traversal
+
+        Returns:
+            bool: True if a circular reference is detected, False otherwise
+        """
+        # If we've already visited this type, we have a cycle
+        if custom_object_type.id in visited:
+            return True
+
+        # Add this type to visited set
+        visited.add(custom_object_type.id)
+
+        # Check all object and multiobject fields in this custom object type
+        for field in custom_object_type.fields.filter(
+            type__in=[
+                CustomFieldTypeChoices.TYPE_OBJECT,
+                CustomFieldTypeChoices.TYPE_MULTIOBJECT,
+            ],
+            related_object_type__isnull=False,
+            related_object_type__app_label=APP_LABEL
+        ):
+
+            # Get the related custom object type directly from the object_type relationship
+            try:
+                next_custom_object_type = CustomObjectType.objects.get(object_type=field.related_object_type)
+            except CustomObjectType.DoesNotExist:
+                continue
+
+            # Recursively check this dependency
+            if self._has_circular_reference(next_custom_object_type, visited):
+                return True
+
+        return False
+
     def serialize(self, value):
         """
         Prepare a value for storage as JSON data.

netbox_custom_objects/tests/test_field_types.py

@@ -736,7 +736,7 @@ def test_self_referential_object_field(self):
             name="parent",
             label="Parent",
             type="object",
-            related_object_type=self.custom_object_type.content_type
+            related_object_type=self.custom_object_type.object_type
         )
         field  # To silence ruff error
 
@@ -759,7 +759,7 @@ def test_self_referential_multiobject_field(self):
             name="children",
             label="Children",
             type="multiobject",
-            related_object_type=self.custom_object_type.content_type
+            related_object_type=self.custom_object_type.object_type
         )
         field  # To silence ruff error
 
@@ -802,7 +802,7 @@ def test_cross_referential_object_field(self):
             name="related_object",
             label="Related Object",
             type="object",
-            related_object_type=second_type.content_type
+            related_object_type=second_type.object_type
         )
         field  # To silence ruff error
 
@@ -834,7 +834,7 @@ def test_cross_referential_multiobject_field(self):
             name="related_objects",
             label="Related Objects",
             type="multiobject",
-            related_object_type=second_type.content_type
+            related_object_type=second_type.object_type
         )
         field  # To silence ruff error
 

netbox_custom_objects/views.py

@@ -203,7 +203,7 @@ def _get_dependent_objects(self, obj):
 
         # Find CustomObjectTypeFields that reference this CustomObjectType
         referencing_fields = CustomObjectTypeField.objects.filter(
-            related_object_type=obj.content_type
+            related_object_type=obj.object_type
         )
 
         # Add the CustomObjectTypeFields that reference this CustomObjectType