Integrate maintenance events with NetBox event system (PR #20 + compatibility fixes)

netbox_maintenance_device/__init__.py

@@ -35,6 +35,36 @@ def ready(self):
         # to avoid issues during initial Django setup and collectstatic operations
         import logging
         logger = logging.getLogger(__name__)
+
+        # Register custom event types
+        try:
+            from netbox.events import (
+                EventType, EVENT_TYPE_KIND_WARNING, EVENT_TYPE_KIND_INFO, EVENT_TYPE_KIND_SUCCESS
+            )
+            from django.utils.translation import gettext
+            from netbox.registry import registry
+
+            # Overwrite registry directly to avoid duplicate registration errors on reload
+            # and use gettext (returning str) instead of gettext_lazy (returning proxy)
+            event_types = [
+                EventType('maintenance_due', gettext('Maintenance due'), kind=EVENT_TYPE_KIND_WARNING),
+                EventType('maintenance_scheduled', gettext('Maintenance scheduled'), kind=EVENT_TYPE_KIND_INFO),
+                EventType('maintenance_completed', gettext('Maintenance completed'), kind=EVENT_TYPE_KIND_SUCCESS),
+            ]
+            for et in event_types:
+                registry['event_types'][et.name] = et
+
+            logger.info("NetBox Maintenance Device: Custom event types registered successfully")
+        except Exception as e:
+            logger.error(f"NetBox Maintenance Device: Custom event types registration failed: {e}")
+
+        # Load system jobs to trigger registration
+        try:
+            import netbox_maintenance_device.jobs  # noqa: F401
+            logger.info("NetBox Maintenance Device: Custom system jobs loaded successfully")
+        except Exception as e:
+            logger.error(f"NetBox Maintenance Device: Custom system jobs load failed: {e}")
+
         logger.info("NetBox Maintenance Device v1.4.1 initialized successfully")
 
 config = MaintenanceDeviceConfig

netbox_maintenance_device/events.py

@@ -0,0 +1,74 @@
+import logging
+from django.conf import settings
+from django.utils.module_loading import import_string
+from core.models import ObjectType
+from extras.events import get_snapshots, serialize_for_event
+from netbox.context import current_request, events_queue
+
+logger = logging.getLogger('netbox.plugins.maintenance_device.events')
+
+
+def fire_event(instance, event_type, request=None):
+    """
+    Fire a custom event for a MaintenancePlan or MaintenanceExecution.
+    If inside a request context (Web UI, API request), it enqueues the event.
+    Otherwise (e.g. management command, background worker), it flushes the event immediately.
+    """
+    if request is None:
+        try:
+            request = current_request.get()
+        except LookupError:
+            request = None
+
+    # Try to get request's event queue
+    queue = None
+    if request is not None:
+        try:
+            queue = events_queue.get()
+        except LookupError:
+            pass
+
+    user = getattr(request, 'user', None) if request else None
+    username = user.username if user else 'system'
+    request_id = getattr(request, 'id', None)
+
+    event_data = {
+        'object_type': ObjectType.objects.get_for_model(instance),
+        'object_id': instance.pk,
+        'object': instance,
+        'event_type': event_type,
+        # Event pipeline consumers (process_event_rules) read event['data']
+        # unconditionally; NetBox 4.6's EventContext fills it lazily, but the
+        # plain-dict path (NetBox <= 4.5 and the immediate-flush branch below)
+        # must carry it eagerly or every matching event rule dies on KeyError.
+        'data': serialize_for_event(instance),
+        'snapshots': get_snapshots(instance, event_type),
+        'request': request,
+        'user': user,
+        'username': username,
+        'request_id': request_id,
+    }
+
+    if queue is not None:
+        app_label = instance._meta.app_label
+        model_name = instance._meta.model_name
+        # Use a key combining model and event type to prevent multiple events
+        # from overwriting each other in the request queue dict.
+        key = f'{app_label}.{model_name}:{instance.pk}:{event_type}'
+
+        try:
+            from extras.events import EventContext
+            queue[key] = EventContext(**event_data)
+        except ImportError:
+            queue[key] = event_data
+
+        logger.debug(f"Enqueued event '{event_type}' for {instance} in request queue")
+    else:
+        # Flush immediately for synchronous processing outside request cycle
+        logger.info(f"Firing event '{event_type}' for {instance} immediately")
+        for name in settings.EVENTS_PIPELINE:
+            try:
+                func = import_string(name)
+                func([event_data])
+            except Exception as e:
+                logger.error(f"Error processing event pipeline {name}: {e}")

netbox_maintenance_device/forms.py

@@ -5,8 +5,9 @@
 
 from dcim.models import Device
 from netbox.filtersets import NetBoxModelFilterSet
-from netbox.forms import NetBoxModelForm
-from utilities.forms.fields import DynamicModelChoiceField
+from netbox.forms import NetBoxModelForm, NetBoxModelFilterSetForm
+from utilities.forms import BOOLEAN_WITH_BLANK_CHOICES
+from utilities.forms.fields import DynamicModelChoiceField, DynamicModelMultipleChoiceField, TagFilterField
 from utilities.forms.rendering import FieldSet
 from virtualization.models import VirtualMachine
 
@@ -138,3 +139,59 @@ class MaintenanceExecutionFilterSet(NetBoxModelFilterSet):
     class Meta:
         model = models.MaintenanceExecution
         fields = ['maintenance_plan', 'status', 'completed']
+
+
+class MaintenancePlanFilterForm(NetBoxModelFilterSetForm):
+    model = models.MaintenancePlan
+    fieldsets = (
+        FieldSet('q', 'filter_id', 'tag'),
+        FieldSet('device', 'virtual_machine', name=_('Target')),
+        FieldSet('maintenance_type', 'frequency_unit', 'is_active', name=_('Attributes')),
+    )
+    device = DynamicModelMultipleChoiceField(
+        queryset=Device.objects.all(),
+        required=False,
+        label=_('Device')
+    )
+    virtual_machine = DynamicModelMultipleChoiceField(
+        queryset=VirtualMachine.objects.all(),
+        required=False,
+        label=_('Virtual machine')
+    )
+    maintenance_type = forms.MultipleChoiceField(
+        choices=models.MaintenancePlan.MAINTENANCE_TYPE_CHOICES,
+        required=False,
+        label=_('Maintenance type')
+    )
+    frequency_unit = forms.MultipleChoiceField(
+        choices=models.MaintenancePlan.FREQUENCY_UNIT_CHOICES,
+        required=False,
+        label=_('Frequency unit')
+    )
+    is_active = forms.NullBooleanField(
+        required=False,
+        label=_('Active'),
+        widget=forms.Select(
+            choices=BOOLEAN_WITH_BLANK_CHOICES
+        )
+    )
+    tag = TagFilterField(model)
+
+
+class MaintenanceExecutionFilterForm(NetBoxModelFilterSetForm):
+    model = models.MaintenanceExecution
+    fieldsets = (
+        FieldSet('q', 'filter_id', 'tag'),
+        FieldSet('maintenance_plan', 'status', name=_('Attributes')),
+    )
+    maintenance_plan = DynamicModelMultipleChoiceField(
+        queryset=models.MaintenancePlan.objects.all(),
+        required=False,
+        label=_('Maintenance plan')
+    )
+    status = forms.MultipleChoiceField(
+        choices=models.MaintenanceExecution.STATUS_CHOICES,
+        required=False,
+        label=_('Status')
+    )
+    tag = TagFilterField(model)

netbox_maintenance_device/jobs.py

@@ -0,0 +1,40 @@
+from core.choices import JobIntervalChoices
+from netbox.jobs import JobRunner, system_job
+from netbox_maintenance_device.models import MaintenancePlan
+from netbox_maintenance_device.events import fire_event
+
+
+@system_job(interval=JobIntervalChoices.INTERVAL_DAILY)
+class CheckMaintenanceJob(JobRunner):
+    class Meta:
+        name = "Check Maintenance Overdue"
+
+    def run(self, *args, **kwargs):
+        self.logger.info("Checking maintenance plans...")
+
+        active_plans = MaintenancePlan.objects.filter(is_active=True)
+        overdue_count = 0
+        notified_count = 0
+
+        for plan in active_plans:
+            if plan.is_overdue():
+                overdue_count += 1
+                next_due = plan.get_next_maintenance_date()
+                if next_due:
+                    next_due_date = next_due.date()
+                    # Check if we already notified for this specific due date
+                    if plan.last_notified_date != next_due_date:
+                        # Check if there is already a scheduled execution to avoid spamming
+                        has_scheduled = plan.executions.filter(status='scheduled').exists()
+                        if not has_scheduled:
+                            self.logger.warning(
+                                f'Plan "{plan}" is overdue (due since {next_due_date})'
+                            )
+                            fire_event(plan, 'maintenance_due')
+                            plan.last_notified_date = next_due_date
+                            plan.save(update_fields=['last_notified_date'])
+                            notified_count += 1
+
+        self.logger.info(
+            f'Check complete. Active plans: {active_plans.count()}, Overdue: {overdue_count}, Notified: {notified_count}'
+        )

netbox_maintenance_device/management/__init__.py

@@ -0,0 +1 @@
+# management package

netbox_maintenance_device/management/commands/__init__.py

@@ -0,0 +1 @@
+# management commands package

netbox_maintenance_device/management/commands/check_maintenance.py

@@ -0,0 +1,38 @@
+from django.core.management.base import BaseCommand
+from django.utils import timezone
+from netbox_maintenance_device.models import MaintenancePlan
+from netbox_maintenance_device.events import fire_event
+
+
+class Command(BaseCommand):
+    help = 'Check for active maintenance plans that are overdue and trigger notifications'
+
+    def handle(self, *args, **options):
+        self.stdout.write('Checking maintenance plans...')
+
+        active_plans = MaintenancePlan.objects.filter(is_active=True)
+        overdue_count = 0
+        notified_count = 0
+
+        for plan in active_plans:
+            if plan.is_overdue():
+                overdue_count += 1
+                next_due = plan.get_next_maintenance_date()
+                if next_due:
+                    next_due_date = next_due.date()
+                    # Check if we already notified for this specific due date
+                    if plan.last_notified_date != next_due_date:
+                        # Check if there is already a scheduled execution to avoid spamming
+                        has_scheduled = plan.executions.filter(status='scheduled').exists()
+                        if not has_scheduled:
+                            self.stdout.write(self.style.WARNING(
+                                f'Plan "{plan}" is overdue (due since {next_due_date})'
+                            ))
+                            fire_event(plan, 'maintenance_due')
+                            plan.last_notified_date = next_due_date
+                            plan.save(update_fields=['last_notified_date'])
+                            notified_count += 1
+
+        self.stdout.write(self.style.SUCCESS(
+            f'Check complete. Active plans: {active_plans.count()}, Overdue: {overdue_count}, Notified: {notified_count}'
+        ))

netbox_maintenance_device/migrations/0004_remove_maintenanceplan_last_executed_and_more.py

@@ -0,0 +1,37 @@
+# Generated by Django 6.0.5 on 2026-05-20 19:56
+
+import taggit.managers
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        # Pinning a concrete extras migration would break installs on NetBox
+        # < 4.6 (extras 0138 ships with 4.6.0); '__latest__' resolves on any
+        # supported version, matching the plugin's other migrations.
+        ("extras", "__latest__"),
+        ("netbox_maintenance_device", "0003_virtual_machine_and_calendar_schedule"),
+    ]
+
+    operations = [
+        migrations.RemoveField(
+            model_name="maintenanceplan",
+            name="last_executed",
+        ),
+        migrations.AddField(
+            model_name="maintenanceexecution",
+            name="tags",
+            field=taggit.managers.TaggableManager(through="extras.TaggedItem", to="extras.Tag"),
+        ),
+        migrations.AddField(
+            model_name="maintenanceplan",
+            name="last_notified_date",
+            field=models.DateField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="maintenanceplan",
+            name="tags",
+            field=taggit.managers.TaggableManager(through="extras.TaggedItem", to="extras.Tag"),
+        ),
+    ]

netbox_maintenance_device/models.py

@@ -84,6 +84,15 @@ class MaintenancePlan(NetBoxModel):
         verbose_name=_('Anchor Date'),
     )
     is_active = models.BooleanField(default=True, verbose_name=_('Active'))
+    last_notified_date = models.DateField(
+        null=True,
+        blank=True,
+        help_text=_(
+            "The last due date for which a notification was sent, to avoid "
+            "duplicate alerts."
+        ),
+        verbose_name=_('Last Notified Date'),
+    )
 
     class Meta:
         ordering = ['device', 'virtual_machine', 'name']
@@ -181,6 +190,8 @@ def get_next_maintenance_date(self):
             completed=True
         ).order_by('-completed_date').first()
         reference = last_execution.completed_date if last_execution else self.created
+        if reference is None:
+            reference = timezone.now()
 
         step = self._step_delta()
 
@@ -253,4 +264,26 @@ def get_absolute_url(self):
     def save(self, *args, **kwargs):
         # Auto-set completed flag based on status
         self.completed = self.status == 'completed'
+
+        is_new = self.pk is None
+        old_status = None
+        if not is_new:
+            try:
+                old_status = MaintenanceExecution.objects.only('status').get(pk=self.pk).status
+            except MaintenanceExecution.DoesNotExist:
+                pass
+
         super().save(*args, **kwargs)
+
+        # Fire events
+        from .events import fire_event
+        if is_new:
+            if self.status == 'scheduled':
+                fire_event(self, 'maintenance_scheduled')
+            elif self.status == 'completed':
+                fire_event(self, 'maintenance_completed')
+        else:
+            if self.status == 'scheduled' and old_status != 'scheduled':
+                fire_event(self, 'maintenance_scheduled')
+            elif self.status == 'completed' and old_status != 'completed':
+                fire_event(self, 'maintenance_completed')