Add background jobs to plugin API

[alehaa]

NetBox version

v3.7.5

Feature type

Change to existing functionality

Proposed functionality

NetBox already includes the functionality to schedule jobs using either the django-rq library or the core.Job model combined with the JobsMixin class. Making the abstract job functionality publicly available to plugins allows decoupling from the backend of django-rq and a consistent experience in NetBox across all functionality. For this I propose:

1. Add netbox.models.JobsMixin to the list of API available model mixins in the documentation. This allows plugins to implement new models with Jobs enabled.

2. Add a new BackgroundJob class to implement the execution of the job's code, i.e. for consistency in calling start, terminate and setting error messages. This class should also be used in existing NetBox functionality to run background jobs for consistency. Below is a sample implementation from a plugin of mine for demonstration purposes.

   import logging
   from abc import ABC, abstractmethod
   
   from rq.timeouts import JobTimeoutException
   
   from core.choices import JobStatusChoices
   from core.models import ObjectType, Job
   
   
   class BackgroundJob(ABC):
       """
       Background Job helper class.
   
       This class handles the execution of a background job. It is responsible for
       maintaining its state, reporting errors, and scheduling recurring jobs.
       """
   
       @classmethod
       @abstractmethod
       def run(cls, *args, **kwargs) -> None:
           """
           Run the job.
   
           A `BackgroundJob` class needs to implement this method to execute all
           commands of the job.
           """
           pass
   
       @classmethod
       def handle(cls, job: Job, *args, **kwargs) -> None:
           """
           Handle the execution of a `BackgroundJob`.
   
           This method is called by the Job Scheduler to handle the execution of
           all job commands. It will maintain the job's metadata and handle errors.
           For periodic jobs, a new job is automatically scheduled using its
           `interval'.
   
   
           :param job: The job to be executed.
           """
           try:
               job.start()
               cls.run(job, *args, **kwargs)
               job.terminate()
   
           except Exception as e:
               job.terminate(status=JobStatusChoices.STATUS_ERRORED, error=repr(e))
               if type(e) == JobTimeoutException:
                   logging.error(e)
   
           # If the executed job is a periodic job, schedule its next execution at
           # the specified interval.
           finally:
               if job.interval:
                   next_schedule_at = (job.scheduled or job.started) + timedelta(
                       minutes=job.interval
                   )
                   cls.enqueue(
                       instance=job.object,
                       name=job.name,
                       user=job.user,
                       schedule_at=next_schedule_at,
                       interval=job.interval,
                       **kwargs,
                   )
   
       @classmethod
       def enqueue(cls, *args, **kwargs) -> None:
           """
           Enqueue a new `BackgroundJob`.
   
           This method is a wrapper of `Job.enqueue` using :py:meth:`handle` as
           function callback. See its documentation for parameters.
           """
           Job.enqueue(cls.handle, *args, **kwargs)

3. Optional: Enhance the core.models.Job.enqueue() method with a queue parameter to schedule jobs for specific queues (e.g. low, high). If not provided the default queue will be used, so there's no change in API compatibility.

4. Optional: Add the ability to schedule system background tasks by plugins, e.g. for periodic synchronization with other systems. Below is a sample implementation from a plugin of mine for demonstration purposes.

   class ScheduledJob(BackgroundJob):
       """
       A periodic `BackgroundJob` used for system tasks.
   
       This class can be used to schedule system background tasks, e.g. to
       periodically perform data synchronization from other systems to NetBox or to
       perform housekeeping tasks.
       """
   
       ENQUEUED_STATUS = [
           JobStatusChoices.STATUS_PENDING,
           JobStatusChoices.STATUS_SCHEDULED,
           JobStatusChoices.STATUS_RUNNING,
       ]
   
       @classmethod
       def schedule(
           cls,
           instance: models.Model,
           name: str = "",
           interval: int = None,
           *args,
           **kwargs,
       ) -> None:
           """
           Schedule a `BackgroundJob`.
   
           This method adds a new `BackgroundJob` to the job queue. If the job
           schedule identified by its `instance` and `name` is already active,
           scheduling a second will be skipped. For additional parameters see
           :py:meth:`Job.enqueue`.
   
           The main use case for this method is to schedule jobs programmatically
           instead of using user events, e.g. to start jobs when the plugin is
           loaded in NetBox instead of when a user performs an event. It can be
           called from the plugin's `ready()` function to safely setup schedules.
   
   
           :param instance: The instance the job is attached to.
           :param name: Name of the job schedule.
           :param interval: Interval in which the job should be scheduled.
           """
           object_type = ObjectType.objects.get_for_model(
               instance,
               for_concrete_model=False,
           )
           job = Job.objects.filter(
               object_type=object_type,
               object_id=instance.pk,
               name=name,
               interval__isnull=(interval is None),
               status__in=cls.ENQUEUED_STATUS,
           ).first()
           if job:
               # If the job parameters haven't changed, don't schedule a new job
               # and reuse the current schedule. Otherwise, delete the existing job
               # and schedule a new job instead.
               if job.interval == interval:
                   return
               job.delete()
   
           cls.enqueue(name=name, interval=interval, *args, **kwargs)
   
   
   class SystemJob(ScheduledJob):
       """
       A `ScheduledJob` not being bound to any particular NetBox object.
   
       This class can be used to schedule system background tasks that are not
       specific to a particular NetBox object, but a general task.
   
       A typical use case for this class is to implement a general synchronization
       of NetBox objects from another system. If the configuration of the other
       system isn't stored in the database, but the NetBox configuration instead,
       there is no object to bind the `Job` object to. This class therefore allows
       unbound jobs to be scheduled for system background tasks.
       """
   
       @classmethod
       def enqueue(cls, *args, **kwargs) -> None:
           kwargs.pop("instance", None)
           super().enqueue(instance=Job(), *args, **kwargs)
   
       @classmethod
       def schedule(cls, *args, **kwargs) -> None:
           kwargs.pop("instance", None)
           super().schedule(instance=Job(), *args, **kwargs)
   
       @classmethod
       def handle(cls, job: Job, *args, **kwargs) -> None:
           # A job requires a related object to be handled, or internal methods
           # will fail. To avoid adding an extra model for this, the existing job
           # object is used as a reference. This is not ideal, but it works for
           # this purpose.
           job.object = job
           job.object_id = None  # Hide changes from UI
   
           super().handle(job, *args, **kwargs)

Use case

1. Plugins get a standardized interface for adding models with jobs enabled using the JobsMixin, just like native NetBox models. This provides a consistent experience.

2. The environment for running background jobs will be standardized, as startup, termination, and error handling will be the same for all jobs. Individual jobs don't have to worry about rescheduling, but can rely on well-tested and managed code.

3. Using the SystemJob interface, plugins could schedule system tasks such as periodic synchronization with other systems (e.g. virtualization clusters) or perform housekeeping. These jobs are usually not bound to a specific NetBox object and currently require either direct access to the django-rq library or use of an external cronjob and management commands.

Database changes

None

External dependencies

None

For the functionality described above I can share my existing code, add test cases and provide a PR for review. Special thanks goes to @wouterdebruijn for sharing his ideas and feedback in the NetBox discussions.

[JonasEinfach]

I think it would be very useful if netbox support easier usage of the jobs for plugins.
Im actually trying to implement a job for a plugin, it's very tricky to find how it works.

A lot of plugins implement a custom solution with the big disadvantage that the jobs not listed in the Jobs Tab.