Merge Feature

.github/workflows/lint-tests.yaml

@@ -0,0 +1,91 @@
+name: Lint and tests
+on:
+  workflow_dispatch:
+  pull_request:
+  push:
+
+concurrency:
+  group: ${{ github.workflow }}
+  cancel-in-progress: false
+
+permissions:
+  contents: write
+  checks: write
+  pull-requests: write
+
+jobs:
+  lint:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.10"
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install .
+          pip install .[dev]
+          pip install .[test]
+      - name: Run ruff
+        run: ruff check
+  tests:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    strategy:
+      matrix:
+        python-version: [ "3.10", "3.11", "3.12" ]
+    services:
+      redis:
+        image: redis
+        ports:
+          - 6379:6379
+      postgres:
+        image: postgres
+        env:
+          POSTGRES_USER: netbox
+          POSTGRES_PASSWORD: netbox
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 5432:5432
+    steps:
+      - name: Checkout netbox-custom-objects
+        uses: actions/checkout@v4
+        with:
+          path: netbox-custom-objects
+      - name: Setup Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Checkout netbox
+        uses: actions/checkout@v4
+        with:
+          repository: "netbox-community/netbox"
+          path: netbox
+          ref: feature
+      - name: Install netbox-custom-objects
+        working-directory: netbox-custom-objects
+        run: |
+          # Include tests directory for test
+          sed -i 's/exclude-package-data/#exclude-package-data/g' pyproject.toml
+          python -m pip install --upgrade pip
+          pip install .
+          pip install .[test]
+      - name: Install dependencies & configure plugin
+        working-directory: netbox
+        run: |
+          ln -s $(pwd)/../netbox-custom-objects/testing/configuration.py netbox/netbox/configuration.py
+
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt -U
+      - name: Run tests
+        working-directory: netbox
+        run: |
+          python netbox/manage.py test netbox_custom_objects.tests --keepdb

netbox_custom_objects/__init__.py

@@ -1,42 +1,159 @@
+import sys
+import warnings
+
+from django.core.exceptions import AppRegistryNotReady
+from django.db import transaction
+from django.db.utils import DatabaseError, OperationalError, ProgrammingError
 from netbox.plugins import PluginConfig
 
 
+def is_running_migration():
+    """
+    Check if the code is currently running during a Django migration.
+    """
+    # Check if 'makemigrations' or 'migrate' command is in sys.argv
+    if any(cmd in sys.argv for cmd in ["makemigrations", "migrate"]):
+        return True
+
+    return False
+
+
+def is_in_clear_cache():
+    """
+    Check if the code is currently being called from Django's clear_cache() method.
+
+    TODO: This is fairly ugly, but in models.CustomObjectType.get_model() we call
+    meta = type() which calls clear_cache on the model which causes a call to
+    get_models() which in-turn calls get_model and therefore recurses.
+
+    This catches the specific case of a recursive call to get_models() from
+    clear_cache() which is the only case we care about, so should be relatively
+    safe.  An alternative should be found for this.
+    """
+    import inspect
+
+    frame = inspect.currentframe()
+    try:
+        # Walk up the call stack to see if we're being called from clear_cache
+        while frame:
+            if (
+                frame.f_code.co_name == "clear_cache"
+                and "django/apps/registry.py" in frame.f_code.co_filename
+            ):
+                return True
+            frame = frame.f_back
+        return False
+    finally:
+        # Clean up the frame reference
+        del frame
+
+
+def check_custom_object_type_table_exists():
+    """
+    Check if the CustomObjectType table exists in the database.
+    Returns True if the table exists, False otherwise.
+    """
+    from .models import CustomObjectType
+
+    try:
+        # Try to query the model - if the table doesn't exist, this will raise an exception
+        # this check and the transaction.atomic() is only required when running tests as the
+        # migration check doesn't work correctly in the test environment
+        with transaction.atomic():
+            # Force immediate execution by using first()
+            CustomObjectType.objects.first()
+        return True
+    except (OperationalError, ProgrammingError, DatabaseError):
+        # Catch database-specific errors (table doesn't exist, permission issues, etc.)
+        return False
+
+
 # Plugin Configuration
 class CustomObjectsPluginConfig(PluginConfig):
     name = "netbox_custom_objects"
     verbose_name = "Custom Objects"
     description = "A plugin to manage custom objects in NetBox"
-    version = "0.1.0"
+    version = "0.2.0"
     base_url = "custom-objects"
-    min_version = "4.2.0"
+    min_version = "4.4.0"
     default_settings = {}
     required_settings = []
     template_extensions = "template_content.template_extensions"
+    _in_get_models = False  # Recursion guard
+
+    def get_model(self, model_name, require_ready=True):
+        try:
+            # if the model is already loaded, return it
+            return super().get_model(model_name, require_ready)
+        except LookupError:
+            try:
+                self.apps.check_apps_ready()
+            except AppRegistryNotReady:
+                raise
+
+        # only do database calls if we are sure the app is ready to avoid
+        # Django warnings
+        if "table" not in model_name.lower() or "model" not in model_name.lower():
+            raise LookupError(
+                "App '%s' doesn't have a '%s' model." % (self.label, model_name)
+            )
+
+        from .models import CustomObjectType
+
+        custom_object_type_id = int(
+            model_name.replace("table", "").replace("model", "")
+        )
+
+        try:
+            obj = CustomObjectType.objects.get(pk=custom_object_type_id)
+        except CustomObjectType.DoesNotExist:
+            raise LookupError(
+                "App '%s' doesn't have a '%s' model." % (self.label, model_name)
+            )
+
+        return obj.get_model()
+
+    def get_models(self, include_auto_created=False, include_swapped=False):
+        """Return all models for this plugin, including custom object type models."""
+
+        # Get the regular Django models first
+        for model in super().get_models(include_auto_created, include_swapped):
+            yield model
+
+        # Prevent recursion
+        if self._in_get_models and is_in_clear_cache():
+            # Skip dynamic model creation if we're in a recursive get_models call
+            return
+
+        self._in_get_models = True
+        try:
+            # Suppress warnings about database calls during model loading
+            with warnings.catch_warnings():
+                warnings.filterwarnings(
+                    "ignore", category=RuntimeWarning, message=".*database.*"
+                )
+                warnings.filterwarnings(
+                    "ignore", category=UserWarning, message=".*database.*"
+                )
+
+                # Skip custom object type model loading if running during migration
+                if (
+                    is_running_migration()
+                    or not check_custom_object_type_table_exists()
+                ):
+                    return
+
+                # Add custom object type models
+                from .models import CustomObjectType
 
-    # def get_model(self, model_name, require_ready=True):
-    #     if require_ready:
-    #         self.apps.check_models_ready()
-    #     else:
-    #         self.apps.check_apps_ready()
-    #
-    #     if model_name.lower() in self.models:
-    #         return self.models[model_name.lower()]
-    #
-    #     from .models import CustomObjectType
-    #     if "table" not in model_name.lower() or "model" not in model_name.lower():
-    #         raise LookupError(
-    #             "App '%s' doesn't have a '%s' model." % (self.label, model_name)
-    #         )
-    #
-    #     custom_object_type_id = int(model_name.replace("table", "").replace("model", ""))
-    #
-    #     try:
-    #         obj = CustomObjectType.objects.get(pk=custom_object_type_id)
-    #     except CustomObjectType.DoesNotExist:
-    #         raise LookupError(
-    #             "App '%s' doesn't have a '%s' model." % (self.label, model_name)
-    #         )
-    #     return obj.get_model()
+                custom_object_types = CustomObjectType.objects.all()
+                for custom_type in custom_object_types:
+                    model = custom_type.get_model()
+                    if model:
+                        yield model
+        finally:
+            # Clean up the recursion guard
+            self._in_get_models = False
 
 
 config = CustomObjectsPluginConfig

netbox_custom_objects/api/serializers.py

@@ -1,3 +1,7 @@
+import logging
+import sys
+
+from core.models import ObjectType
 from django.contrib.contenttypes.models import ContentType
 from extras.choices import CustomFieldTypeChoices
 from netbox.api.serializers import NetBoxModelSerializer
@@ -6,7 +10,11 @@
 from rest_framework.reverse import reverse
 
 from netbox_custom_objects import field_types
-from netbox_custom_objects.models import CustomObject, CustomObjectType, CustomObjectTypeField
+from netbox_custom_objects.models import (CustomObject, CustomObjectType,
+                                          CustomObjectTypeField)
+
+logger = logging.getLogger('netbox_custom_objects.api.serializers')
+
 
 __all__ = (
     "CustomObjectTypeSerializer",
@@ -59,10 +67,10 @@ def validate(self, attrs):
             CustomFieldTypeChoices.TYPE_MULTIOBJECT,
         ]:
             try:
-                attrs["related_object_type"] = ContentType.objects.get(
+                attrs["related_object_type"] = ObjectType.objects.get(
                     app_label=app_label, model=model
                 )
-            except ContentType.DoesNotExist:
+            except ObjectType.DoesNotExist:
                 raise ValidationError(
                     "Must provide valid app_label and model for object field type."
                 )
@@ -197,17 +205,24 @@ def get_field_data(self, obj):
 
 def get_serializer_class(model):
     model_fields = model.custom_object_type.fields.all()
+
+    # Create field list including all necessary fields
+    base_fields = ["id", "url", "display", "created", "last_updated", "tags"]
+    custom_field_names = [field.name for field in model_fields]
+    all_fields = base_fields + custom_field_names
+
     meta = type(
         "Meta",
         (),
         {
             "model": model,
-            "fields": "__all__",
+            "fields": all_fields,
+            "brief_fields": ("id", "url", "display"),
         },
     )
 
     def get_url(self, obj):
-        # Unsaved objects will not yet have a valid URL.
+        """Generate the API URL for this object"""
         if hasattr(obj, "pk") and obj.pk in (None, ""):
             return None
 
@@ -221,24 +236,38 @@ def get_url(self, obj):
         format = self.context.get("format")
         return reverse(view_name, kwargs=kwargs, request=request, format=format)
 
+    def get_display(self, obj):
+        """Get display representation of the object"""
+        return str(obj)
+
+    # Create basic attributes for the serializer
     attrs = {
         "Meta": meta,
-        "__module__": "database.serializers",
+        "__module__": "netbox_custom_objects.api.serializers",
         "url": serializers.SerializerMethodField(),
         "get_url": get_url,
+        "display": serializers.SerializerMethodField(),
+        "get_display": get_display,
     }
 
     for field in model_fields:
         field_type = field_types.FIELD_TYPE_CLASS[field.type]()
         try:
             attrs[field.name] = field_type.get_serializer_field(field)
         except NotImplementedError:
-            print(f"serializer: {field.name} field is not implemented; using a default serializer field")
+            logger.debug(
+                "serializer: {} field is not implemented; using a default serializer field".format(field.name)
+            )
 
+    serializer_name = f"{model._meta.object_name}Serializer"
     serializer = type(
-        f"{model._meta.object_name}Serializer",
-        (serializers.ModelSerializer,),
+        serializer_name,
+        (NetBoxModelSerializer,),
         attrs,
     )
 
+    # Register the serializer in the current module so NetBox can find it
+    current_module = sys.modules[__name__]
+    setattr(current_module, serializer_name, serializer)
+
     return serializer