Closes: #391 (Phase 2) - post_migrate auto-heal for mixin column drift

netbox_custom_objects/__init__.py

@@ -32,6 +32,42 @@ def _migration_finished(sender, **kwargs):
     _migrations_checked = None
 
 
+# Module-level flag so the heal runs at most once per process invocation even
+# though post_migrate fires once per installed app.
+_heal_ran = False
+
+
+def _heal_mixin_columns(sender, **kwargs):
+    """
+    post_migrate signal handler: detect and apply mixin column drift.
+
+    Fires after every 'manage.py migrate' run (once per installed app).  The
+    module-level _heal_ran flag ensures the actual work happens only once per
+    process so the cost is negligible on normal server starts where no
+    migrations run.
+
+    Skipped during makemigrations and collectstatic (DB may be unavailable or
+    in an inconsistent state for our purposes).
+    """
+    global _heal_ran
+    if _heal_ran:
+        return
+
+    if any(cmd in sys.argv for cmd in ("makemigrations", "collectstatic")):
+        return
+
+    _heal_ran = True
+
+    try:
+        from netbox_custom_objects.mixin_migration import heal_all_cots  # noqa: PLC0415
+        heal_all_cots(verbosity=kwargs.get("verbosity", 1))
+    except Exception:
+        import logging  # noqa: PLC0415
+        logging.getLogger(__name__).exception(
+            "upgrade_custom_objects: unexpected error during mixin drift check"
+        )
+
+
 def _patch_object_selector_view():
     """
     Patch ObjectSelectorView to support dynamically-generated custom object models.
@@ -180,6 +216,9 @@ def ready(self):
         pre_migrate.connect(_migration_started)
         post_migrate.connect(_migration_finished)
 
+        # Heal mixin column drift after every migrate run (issue #391 Phase 2)
+        post_migrate.connect(_heal_mixin_columns)
+
         # Patch ObjectSelectorView to support dynamically-generated custom object models
         _patch_object_selector_view()
 

netbox_custom_objects/management/__init__.py

@@ -0,0 +1 @@
+# Management commands for netbox_custom_objects

netbox_custom_objects/management/commands/__init__.py

@@ -0,0 +1 @@
+# Management commands for netbox_custom_objects

netbox_custom_objects/management/commands/upgrade_custom_objects.py

@@ -0,0 +1,126 @@
+"""
+management command: upgrade_custom_objects
+
+Checks all Custom Object Type tables for mixin column drift and applies safe
+fixes.  Intended as an explicit escape hatch alongside the automatic
+post_migrate signal handler (issue #391).
+
+Usage examples
+--------------
+    # Check and fix all COTs
+    manage.py upgrade_custom_objects
+
+    # Preview changes without touching the DB
+    manage.py upgrade_custom_objects --dry-run
+
+    # Operate on a single COT (by name or numeric ID)
+    manage.py upgrade_custom_objects --cot my_device
+    manage.py upgrade_custom_objects --cot 7 --dry-run
+"""
+
+from django.core.management.base import BaseCommand, CommandError
+
+from netbox_custom_objects.mixin_migration import heal_cot
+
+
+class Command(BaseCommand):
+    help = (
+        "Detect and apply mixin column drift for Custom Object Type tables. "
+        "New columns contributed by the CustomObject base class (e.g. from a "
+        "NetBox upgrade) are added automatically when nullable or defaulted. "
+        "Non-nullable columns without defaults and column removals are reported "
+        "but never applied automatically."
+    )
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--dry-run",
+            action="store_true",
+            help="Report what would change without making any DB modifications.",
+        )
+        parser.add_argument(
+            "--cot",
+            metavar="NAME_OR_ID",
+            help="Limit to a single Custom Object Type (name or numeric ID).",
+        )
+
+    def handle(self, *args, **options):
+        dry_run = options["dry_run"]
+        cot_filter = options.get("cot")
+        verbosity = options["verbosity"]
+
+        if dry_run:
+            self.stdout.write(self.style.WARNING("DRY RUN — no changes will be made.\n"))
+
+        if cot_filter:
+            from netbox_custom_objects.models import CustomObjectType  # noqa: PLC0415
+            try:
+                if cot_filter.isdigit():
+                    cot = CustomObjectType.objects.get(pk=int(cot_filter))
+                else:
+                    cot = CustomObjectType.objects.get(name=cot_filter)
+            except CustomObjectType.DoesNotExist:
+                raise CommandError(f"No Custom Object Type found: {cot_filter!r}")
+
+            result = heal_cot(cot, verbosity=verbosity, dry_run=dry_run)
+            self._print_cot_result(cot.name, result, dry_run)
+        else:
+            from netbox_custom_objects.models import CustomObjectType  # noqa: PLC0415
+            cots = list(CustomObjectType.objects.all())
+            total = len(cots)
+            healed = warnings = 0
+            for cot in cots:
+                result = heal_cot(cot, verbosity=verbosity, dry_run=dry_run)
+                self._print_cot_result(cot.name, result, dry_run)
+                if result["added"]:
+                    healed += 1
+                warnings += len(result["warned"])
+            self._print_summary(
+                {"total": total, "healed": healed, "warnings": warnings},
+                dry_run,
+            )
+
+    # ------------------------------------------------------------------
+    # Output helpers
+    # ------------------------------------------------------------------
+
+    def _print_cot_result(self, cot_name, result, dry_run):
+        added = result["added"]
+        warned = result["warned"]
+
+        if not added and not warned:
+            self.stdout.write(
+                self.style.SUCCESS(f"COT {cot_name!r}: no drift detected.")
+            )
+            return
+
+        tag = " [DRY RUN]" if dry_run else ""
+        for field_name in added:
+            self.stdout.write(
+                self.style.SUCCESS(f"  {tag} + Added column: {field_name}")
+            )
+        for entry in warned:
+            self.stdout.write(
+                self.style.WARNING(f"  ! {entry['message']}")
+            )
+
+    def _print_summary(self, summary, dry_run):
+        tag = " (dry run)" if dry_run else ""
+        if summary["healed"] == 0 and summary["warnings"] == 0:
+            self.stdout.write(
+                self.style.SUCCESS(
+                    f"All {summary['total']} COT table(s) are up to date{tag}."
+                )
+            )
+        else:
+            self.stdout.write(
+                f"{summary['total']} COT(s) checked{tag}: "
+                f"{summary['healed']} healed, "
+                f"{summary['warnings']} warning(s)."
+            )
+            if summary["warnings"]:
+                self.stdout.write(
+                    self.style.WARNING(
+                        "Run with -v 2 or check the application log for warning details."
+                    )
+                )