Add new Scheduled Task to netbox plugin

[hamza-hadda]

hey
I'm just get started with netbox , I'm developing a plugin in netbox and I want to add a daily task which run everyday , but I didn't find any configuration of a scheduled tasks for that , can you help to do that if you have any ideas

[candlerb]

https://docs.netbox.dev/en/stable/features/background-jobs/#scheduled-jobs

[hamza-hadda]

I didn't find anyway to schedule a daily job

[candlerb]

From the documentation link I posted:

NetBox plugins can enqueue their own background tasks. This is accomplished using the Job model

So, did you try creating and saving an instance of the Job model to represent your task? If this isn't working for you, then post the contents of your Job model instance.

As for making it daily, follow the link to the Job model where it says:

Interval

The interval (in minutes) at which a scheduled job should re-execute.

[wouterdebruijn]

This is done through the Job model as mentioned above, but the implementation of enqueuing a function isn't well documented. I have created the following example by reading the implementation of Netbox's own jobs Like the script implementation.

If you want to run the job at an interval since Netbox Startup:

Example of running the job on startup

# example_plugin/__init__.py
from datetime import datetime, timezone
from extras.plugins import PluginConfig

class ExamplePlugin(PluginConfig):
  name = 'example_plugin'
  version = '0.1'

  def ready(self):
        super().ready()

        from core.models import Job
        from .models import ExampleModel
        from .jobs import sync_discrepancies, ENQUEUED_STATUS

        if Job.objects.filter(
            name="Running sync for example Model",
            status__in=ENQUEUED_STATUS,
            scheduled__gt=datetime.now(timezone.utc),
        ).exists():
            return

        Job.enqueue(
            sync_discrepancies,
            instance=ExampleModel.objects.first(),
            name="Running sync for example Model",
            user=None,
            interval=60,
            schedule_at=datetime.now(timezone.utc),
        )

The implementation of the job function

# example_plugin/jobs.py
from datetime import timedelta
from core.models import Job
import logging
from time import tzname
from core.choices import JobStatusChoices

logger = logging.getLogger(__name__)

ENQUEUED_STATUS = [
    JobStatusChoices.STATUS_PENDING,
    JobStatusChoices.STATUS_RUNNING,
    JobStatusChoices.STATUS_SCHEDULED,
]

# https://github.com/netbox-community/netbox/blob/d2fee886001e4abbcb1b4d9ed3fd32521c820be9/netbox/extras/reports.py#L30
def example_job(job: Job, *args, **kwargs):
  try:
    job.start()
    print("We are running the job!")
    job.terminate()

  except Exception as e:
    job.terminate(status=JobStatusChoices.STATUS_ERRORED, error=repr(e))

  finally:
    
    if job.interval:
      new_scheduled_time = job.scheduled + timedelta(minutes=job.interval)
      job.enqueue(
        example_job,
        instance=job.object,
        name=job.name,
        user=job.user,
        schedule_at=new_scheduled_time,
        interval=job.interval,
      )

[alehaa]

This is wonderful, thanks! Do you know if by any chance one can schedule a job without an existing module instance? I'd like to schedule a background task for synchronizing data in general, not related to a specific object.

[wouterdebruijn]

I tested it with a Null object and Netbox didn't like that if I remember correctly. If it is just a background job synchronizing some data, it might be easiest to do a Device.objects.first(), or any other default Netbox object that will always exist like Site or maybe even User.

[wouterdebruijn]

Just as a heads up, the def ready(self): plugin implementation I mentioned above isn't perfect yet. And I haven't tested this code in a production environment yet. def ready(self): will function a lot better in production compared to development. See this stackoverflow post for some more information.

[alehaa]

I checked using a simple dummy model for this.

from django.db import models


class DummyModel(models.Model):
    class Meta:
        managed = False

Job.enqueue(
    sync,
    instance=DummyModel(),
    name="My sync job",
    user=None,
    interval=60,
    schedule_at=datetime.now(timezone.utc),
)

This seems to make NetBox code happy. I'll give it a try after further testing.

[alehaa]

It turns out that using an additional model introduces its own problems, so I ended up using the already existing job object. For those interested in the full code, I've created feature request #15692, which includes all three classes that provide different background job types.