Using RQ Tasks with Netbox scripts for Asynchronous Execution and Retry Logic

The scripting functionality in Netbox is an easy way to add simple functionality and integrations without having to resort to real django apps. However, by default these scripts run synchronously and will not retry on error. You can, though, use the python-rq integration that comes with netbox out of the box.

It requires you to define the task separately in a separate module (else the rq worker will fail to properly resolve/load it). E.g. in your scripts folder, create a tasks.py with a sample task as follows

import random

def my_task(a, b):
    if random.randint(0, 4) != 3:
        raise Exception("Oopsie")

This simple tasks will randomly fail 25% of the time, but we want netbox to retry it up to 10 times.

Add the following script to scripts/sample.py

from extras.scripts import Script
from django_rq import get_queue
from rq.job import Retry

from scripts.tasks import my_task

class MyScript(Script):
    def run(self, data, commit):
        self.log_debug("Start of script")

        queue = get_queue()
        retry = Retry(max=5, interval=10)
        queue.enqueue(my_task, 42, 31337, retry=retry)

        self.log_debug("End of script")
        return "ok"

This will try the task at most 5 times until it succeeds with a 10 second interval. Consult the python-rq documentation to see how to tweak these values to your needs.