`nautobot.apps.jobs` ¶

Nautobot Jobs API.

`nautobot.apps.jobs.BaseJob` ¶

Bases: Task

Base model for jobs.

Users can subclass this directly if they want to provide their own base class for implementing multiple jobs with shared functionality; if no such sharing is required, use Job class instead.

Jobs must define at minimum a run method.

Source code in nautobot/extras/jobs.py

class BaseJob(Task):
    """Base model for jobs.

    Users can subclass this directly if they want to provide their own base class for implementing multiple jobs
    with shared functionality; if no such sharing is required, use Job class instead.

    Jobs must define at minimum a run method.
    """

    class Meta:
        """
        Metaclass attributes - subclasses can define any or all of the following attributes:

        - name (str)
        - description (str)
        - hidden (bool)
        - field_order (list)
        - approval_required (bool)
        - soft_time_limit (int)
        - time_limit (int)
        - has_sensitive_variables (bool)
        - task_queues (list)
        """

    def __init__(self):
        self.logger = get_task_logger(self.__module__)

    def __call__(self, *args, **kwargs):
        # Attempt to resolve serialized data back into original form by creating querysets or model instances
        # If we fail to find any objects, we consider this a job execution error, and fail.
        # This might happen when a job sits on the queue for a while (i.e. scheduled) and data has changed
        # or it might be bad input from an API request, or manual execution.
        try:
            deserialized_kwargs = self.deserialize_data(kwargs)
        except Exception as err:
            raise RunJobTaskFailed("Error initializing job") from err
        if isinstance(self, JobHookReceiver):
            change_context = ObjectChangeEventContextChoices.CONTEXT_JOB_HOOK
        else:
            change_context = ObjectChangeEventContextChoices.CONTEXT_JOB

        with web_request_context(user=self.user, context_detail=self.class_path, context=change_context):
            if self.celery_kwargs.get("nautobot_job_profile", False) is True:
                import cProfile

                # TODO: This should probably be available as a file download rather than dumped to the hard drive.
                # Pending this: https://github.com/nautobot/nautobot/issues/3352
                profiling_path = f"{tempfile.gettempdir()}/nautobot-jobresult-{self.job_result.id}.pstats"
                self.logger.info(
                    "Writing profiling information to %s.", profiling_path, extra={"grouping": "initialization"}
                )

                with cProfile.Profile() as pr:
                    try:
                        output = self.run(*args, **deserialized_kwargs)
                    except Exception as err:
                        pr.dump_stats(profiling_path)
                        raise err
                    else:
                        pr.dump_stats(profiling_path)
                        return output
            else:
                return self.run(*args, **deserialized_kwargs)

    def __str__(self):
        return str(self.name)

    # See https://github.com/PyCQA/pylint-django/issues/240 for why we have a pylint disable on each classproperty below

    # TODO(jathan): Could be interesting for custom stuff when the Job is
    # enabled in the database and then therefore registered in Celery
    @classmethod
    def on_bound(cls, app):
        """Called when the task is bound to an app.

        Note:
            This class method can be defined to do additional actions when
            the task class is bound to an app.
        """

    # TODO(jathan): Could be interesting for showing the Job's class path as the
    # shadow name vs. the Celery task_name?
    def shadow_name(self, args, kwargs, options):
        """Override for custom task name in worker logs/monitoring.

        Example:
                from celery.utils.imports import qualname

                def shadow_name(task, args, kwargs, options):
                    return qualname(args[0])

                @app.task(shadow_name=shadow_name, serializer='pickle')
                def apply_function_async(fun, *args, **kwargs):
                    return fun(*args, **kwargs)

        Arguments:
            args (Tuple): Task positional arguments.
            kwargs (Dict): Task keyword arguments.
            options (Dict): Task execution options.
        """

    def before_start(self, task_id, args, kwargs):
        """Handler called before the task starts.

        Arguments:
            task_id (str): Unique id of the task to execute.
            args (Tuple): Original arguments for the task to execute.
            kwargs (Dict): Original keyword arguments for the task to execute.

        Returns:
            (None): The return value of this handler is ignored.
        """
        self.clear_cache()

        try:
            self.job_result
        except ObjectDoesNotExist as err:
            raise RunJobTaskFailed(f"Unable to find associated job result for job {task_id}") from err

        try:
            self.job_model
        except ObjectDoesNotExist as err:
            raise RunJobTaskFailed(f"Unable to find associated job model for job {task_id}") from err

        if not self.job_model.enabled:
            self.logger.error(
                "Job %s is not enabled to be run!",
                self.job_model,
                extra={"object": self.job_model, "grouping": "initialization"},
            )
            raise RunJobTaskFailed(f"Job {self.job_model} is not enabled to be run!")

        soft_time_limit = self.job_model.soft_time_limit or settings.CELERY_TASK_SOFT_TIME_LIMIT
        time_limit = self.job_model.time_limit or settings.CELERY_TASK_TIME_LIMIT
        if time_limit <= soft_time_limit:
            self.logger.warning(
                "The hard time limit of %s seconds is less than "
                "or equal to the soft time limit of %s seconds. "
                "This job will fail silently after %s seconds.",
                time_limit,
                soft_time_limit,
                time_limit,
                extra={"grouping": "initialization"},
            )

        self.logger.info("Running job", extra={"grouping": "initialization"})

    def run(self, *args, **kwargs):
        """
        Method invoked when this Job is run.
        """
        raise NotImplementedError("Jobs must define the run method.")

    def on_success(self, retval, task_id, args, kwargs):
        """Success handler.

        Run by the worker if the task executes successfully.

        Arguments:
            retval (Any): The return value of the task.
            task_id (str): Unique id of the executed task.
            args (Tuple): Original arguments for the executed task.
            kwargs (Dict): Original keyword arguments for the executed task.

        Returns:
            (None): The return value of this handler is ignored.
        """

    def on_retry(self, exc, task_id, args, kwargs, einfo):
        """Retry handler.

        This is run by the worker when the task is to be retried.

        Arguments:
            exc (Exception): The exception sent to :meth:`retry`.
            task_id (str): Unique id of the retried task.
            args (Tuple): Original arguments for the retried task.
            kwargs (Dict): Original keyword arguments for the retried task.
            einfo (~billiard.einfo.ExceptionInfo): Exception information.

        Returns:
            (None): The return value of this handler is ignored.
        """

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        """Error handler.

        This is run by the worker when the task fails.

        Arguments:
            exc (Exception): The exception raised by the task.
            task_id (str): Unique id of the failed task.
            args (Tuple): Original arguments for the task that failed.
            kwargs (Dict): Original keyword arguments for the task that failed.
            einfo (~billiard.einfo.ExceptionInfo): Exception information.

        Returns:
            (None): The return value of this handler is ignored.
        """

    def after_return(self, status, retval, task_id, args, kwargs, einfo):
        """
        Handler called after the task returns.

        Parameters
            status - Current task state.
            retval - Task return value/exception.
            task_id - Unique id of the task.
            args - Original arguments for the task that returned.
            kwargs - Original keyword arguments for the task that returned.

        Keyword Arguments
            einfo - ExceptionInfo instance, containing the traceback (if any).

        Returns:
            (None): The return value of this handler is ignored.
        """

        # Cleanup FileProxy objects
        file_fields = list(self._get_file_vars())
        file_ids = [kwargs[f] for f in file_fields]
        if file_ids:
            self.delete_files(*file_ids)

        self.logger.info("Job completed", extra={"grouping": "post_run"})

        # TODO(gary): document this in job author docs
        # Super.after_return must be called for chords to function properly
        super().after_return(status, retval, task_id, args, kwargs, einfo=einfo)

    def apply(
        self,
        args=None,
        kwargs=None,
        link=None,
        link_error=None,
        task_id=None,
        retries=None,
        throw=None,
        logfile=None,
        loglevel=None,
        headers=None,
        **options,
    ):
        """Fix celery's apply method to propagate options to the task result"""
        # trace imports Task, so need to import inline.
        from celery.app.trace import build_tracer

        app = self._get_app()
        args = args or ()
        kwargs = kwargs or {}
        task_id = task_id or uuid()
        retries = retries or 0
        if throw is None:
            throw = app.conf.task_eager_propagates

        # Make sure we get the task instance, not class.
        task = app._tasks[self.name]

        request = {
            "id": task_id,
            "retries": retries,
            "is_eager": True,
            "logfile": logfile,
            "loglevel": loglevel or 0,
            "hostname": gethostname(),
            "callbacks": maybe_list(link),
            "errbacks": maybe_list(link_error),
            "headers": headers,
            "ignore_result": options.get("ignore_result", False),
            "delivery_info": {
                "is_eager": True,
                "exchange": options.get("exchange"),
                "routing_key": options.get("routing_key"),
                "priority": options.get("priority"),
            },
            "properties": options,  # one line fix to overloaded method
        }
        if "stamped_headers" in options:
            request["stamped_headers"] = maybe_list(options["stamped_headers"])
            request["stamps"] = {header: maybe_list(options.get(header, [])) for header in request["stamped_headers"]}

        tb = None
        tracer = build_tracer(
            task.name,
            task,
            eager=True,
            propagate=throw,
            app=self._get_app(),
        )
        ret = tracer(task_id, args, kwargs, request)
        retval = ret.retval
        if isinstance(retval, ExceptionInfo):
            retval, tb = retval.exception, retval.traceback
            if isinstance(retval, ExceptionWithTraceback):
                retval = retval.exc
        if isinstance(retval, Retry) and retval.sig is not None:
            return retval.sig.apply(retries=retries + 1)
        state = states.SUCCESS if ret.info is None else ret.info.state
        return EagerResult(task_id, retval, state, traceback=tb)

    @final
    @classproperty
    def file_path(cls) -> str:  # pylint: disable=no-self-argument
        return inspect.getfile(cls)

    @final
    @classproperty
    def class_path(cls) -> str:  # pylint: disable=no-self-argument
        """
        Unique identifier of a specific Job class, in the form <module_name>.<ClassName>.

        Examples:
        my_script.MyScript - Local Job
        nautobot.core.jobs.MySystemJob - System Job
        my_plugin.jobs.MyPluginJob - App-provided Job
        git_repository.jobs.myjob.MyJob - GitRepository Job
        """
        return f"{cls.__module__}.{cls.__name__}"

    @final
    @classproperty
    def class_path_dotted(cls) -> str:  # pylint: disable=no-self-argument
        """
        Dotted class_path, suitable for use in things like Python logger names.

        Deprecated as of Nautobot 2.0: just use .class_path instead.
        """
        return cls.class_path

    @final
    @classproperty
    def class_path_js_escaped(cls) -> str:  # pylint: disable=no-self-argument
        """
        Escape various characters so that the class_path can be used as a jQuery selector.
        """
        return cls.class_path.replace(".", r"\.")

    @final
    @classproperty
    def grouping(cls) -> str:  # pylint: disable=no-self-argument
        module = inspect.getmodule(cls)
        return getattr(module, "name", module.__name__)

    @final
    @classmethod
    def _get_meta_attr_and_assert_type(cls, attr_name, default, expected_type):
        result = getattr(cls.Meta, attr_name, default)
        if not isinstance(result, expected_type):
            raise TypeError(f"Meta.{attr_name} should be {expected_type}, not {type(result)}")
        return result

    @final
    @classproperty
    def name(cls) -> str:  # pylint: disable=no-self-argument
        return cls._get_meta_attr_and_assert_type("name", cls.__name__, expected_type=str)

    @final
    @classproperty
    def description(cls) -> str:  # pylint: disable=no-self-argument
        return dedent(cls._get_meta_attr_and_assert_type("description", "", expected_type=str)).strip()

    @final
    @classproperty
    def description_first_line(cls) -> str:  # pylint: disable=no-self-argument
        if cls.description:  # pylint: disable=using-constant-test
            return cls.description.splitlines()[0]
        return ""

    @final
    @classproperty
    def dryrun_default(cls) -> bool:  # pylint: disable=no-self-argument
        return cls._get_meta_attr_and_assert_type("dryrun_default", False, expected_type=bool)

    @final
    @classproperty
    def hidden(cls) -> bool:  # pylint: disable=no-self-argument
        return cls._get_meta_attr_and_assert_type("hidden", False, expected_type=bool)

    @final
    @classproperty
    def field_order(cls):  # pylint: disable=no-self-argument
        return cls._get_meta_attr_and_assert_type("field_order", [], expected_type=(list, tuple))

    @final
    @classproperty
    def read_only(cls) -> bool:  # pylint: disable=no-self-argument
        return cls._get_meta_attr_and_assert_type("read_only", False, expected_type=bool)

    @final
    @classproperty
    def approval_required(cls) -> bool:  # pylint: disable=no-self-argument
        return cls._get_meta_attr_and_assert_type("approval_required", False, expected_type=bool)

    @final
    @classproperty
    def soft_time_limit(cls) -> int:  # pylint: disable=no-self-argument
        return cls._get_meta_attr_and_assert_type("soft_time_limit", 0, expected_type=int)

    @final
    @classproperty
    def time_limit(cls) -> int:  # pylint: disable=no-self-argument
        return cls._get_meta_attr_and_assert_type("time_limit", 0, expected_type=int)

    @final
    @classproperty
    def has_sensitive_variables(cls) -> bool:  # pylint: disable=no-self-argument
        return cls._get_meta_attr_and_assert_type("has_sensitive_variables", True, expected_type=bool)

    @final
    @classproperty
    def supports_dryrun(cls) -> bool:  # pylint: disable=no-self-argument
        return isinstance(getattr(cls, "dryrun", None), DryRunVar)

    @final
    @classproperty
    def task_queues(cls) -> list:  # pylint: disable=no-self-argument
        return cls._get_meta_attr_and_assert_type("task_queues", [], expected_type=(list, tuple))

    @final
    @classproperty
    def properties_dict(cls) -> dict:  # pylint: disable=no-self-argument
        """
        Return all relevant classproperties as a dict.

        Used for convenient rendering into job_edit.html via the `json_script` template tag.
        """
        return {
            "name": cls.name,
            "grouping": cls.grouping,
            "description": cls.description,
            "approval_required": cls.approval_required,
            "hidden": cls.hidden,
            "soft_time_limit": cls.soft_time_limit,
            "time_limit": cls.time_limit,
            "has_sensitive_variables": cls.has_sensitive_variables,
            "task_queues": cls.task_queues,
        }

    @final
    @classproperty
    def registered_name(cls) -> str:  # pylint: disable=no-self-argument
        return f"{cls.__module__}.{cls.__name__}"

    @classmethod
    def _get_vars(cls):
        """
        Return dictionary of ScriptVariable attributes defined on this class and any base classes to the top of the inheritance chain.
        The variables are sorted in the order that they were defined, with variables defined on base classes appearing before subclass variables.
        """
        cls_vars = {}
        # get list of base classes, including cls, in reverse method resolution order: [BaseJob, Job, cls]
        base_classes = reversed(inspect.getmro(cls))
        attr_names = [name for base in base_classes for name in base.__dict__.keys()]
        for name in attr_names:
            attr_class = getattr(cls, name, None).__class__
            if name not in cls_vars and issubclass(attr_class, ScriptVariable):
                cls_vars[name] = getattr(cls, name)

        return cls_vars

    @classmethod
    def _get_file_vars(cls):
        """Return an ordered dict of FileVar fields."""
        cls_vars = cls._get_vars()
        file_vars = OrderedDict()
        for name, attr in cls_vars.items():
            if isinstance(attr, FileVar):
                file_vars[name] = attr

        return file_vars

    @classmethod
    def as_form_class(cls):
        """
        Dynamically generate a Django form class corresponding to the variables in this Job.

        In most cases you should use `.as_form()` instead of calling this method directly.
        """
        fields = {name: var.as_field() for name, var in cls._get_vars().items()}
        return type("JobForm", (JobForm,), fields)

    @classmethod
    def as_form(cls, data=None, files=None, initial=None, approval_view=False):
        """
        Return a Django form suitable for populating the context data required to run this Job.

        `approval_view` will disable all fields from modification and is used to display the form
        during a approval review workflow.
        """

        form = cls.as_form_class()(data, files, initial=initial)

        try:
            job_model = JobModel.objects.get_for_class_path(cls.class_path)
            dryrun_default = job_model.dryrun_default if job_model.dryrun_default_override else cls.dryrun_default
            task_queues = job_model.task_queues if job_model.task_queues_override else cls.task_queues
        except JobModel.DoesNotExist:
            logger.error("No Job instance found in the database corresponding to %s", cls.class_path)
            dryrun_default = cls.dryrun_default
            task_queues = cls.task_queues

        # Update task queue choices
        form.fields["_task_queue"].choices = task_queues_as_choices(task_queues)

        if cls.supports_dryrun and (not initial or "dryrun" not in initial):
            # Set initial "dryrun" checkbox state based on the Meta parameter
            form.fields["dryrun"].initial = dryrun_default
        if not settings.DEBUG:
            form.fields["_profile"].widget = forms.HiddenInput()

        # https://github.com/PyCQA/pylint/issues/3484
        if cls.field_order:  # pylint: disable=using-constant-test
            form.order_fields(cls.field_order)

        if approval_view:
            # Set `disabled=True` on all fields
            for _, field in form.fields.items():
                field.disabled = True

        return form

    def clear_cache(self):
        """
        Clear all cached properties on this instance without accessing them. This is required because
        celery reuses task instances for multiple runs.
        """
        try:
            del self.celery_kwargs
        except AttributeError:
            pass
        try:
            del self.job_result
        except AttributeError:
            pass
        try:
            del self.job_model
        except AttributeError:
            pass

    @functools.cached_property
    def job_model(self):
        return JobModel.objects.get(module_name=self.__module__, job_class_name=self.__name__)

    @functools.cached_property
    def job_result(self):
        return JobResult.objects.get(id=self.request.id)

    @functools.cached_property
    def celery_kwargs(self):
        return self.job_result.celery_kwargs or {}

    @property
    def user(self):
        return getattr(self.job_result, "user", None)

    @staticmethod
    def serialize_data(data):
        """
        This method parses input data (from JobForm usually) and returns a dict which is safe to serialize

        Here we convert the QuerySet of a MultiObjectVar to a list of the pk's and the model instance
        of an ObjectVar into the pk value.

        These are converted back during job execution.
        """

        return_data = {}
        for field_name, value in data.items():
            # MultiObjectVar
            if isinstance(value, QuerySet):
                return_data[field_name] = list(value.values_list("pk", flat=True))
            # ObjectVar
            elif isinstance(value, Model):
                return_data[field_name] = value.pk
            # FileVar (Save each FileVar as a FileProxy)
            elif isinstance(value, InMemoryUploadedFile):
                return_data[field_name] = BaseJob.save_file(value)
            # IPAddressVar, IPAddressWithMaskVar, IPNetworkVar
            elif isinstance(value, netaddr.ip.BaseIP):
                return_data[field_name] = str(value)
            # Everything else...
            else:
                return_data[field_name] = value

        return return_data

    # TODO: can the deserialize_data logic be moved to NautobotKombuJSONEncoder?
    @classmethod
    def deserialize_data(cls, data):
        """
        Given data input for a job execution, deserialize it by resolving object references using defined variables.

        This converts a list of pk's back into a QuerySet for MultiObjectVar instances and single pk values into
        model instances for ObjectVar.

        Note that when resolving querysets or model instances by their PK, we do not catch DoesNotExist
        exceptions here, we leave it up the caller to handle those cases. The normal job execution code
        path would consider this a failure of the job execution, as described in `nautobot.extras.jobs.run_job`.
        """
        cls_vars = cls._get_vars()
        return_data = {}

        if not isinstance(data, dict):
            raise TypeError("Data should be a dictionary.")

        for field_name, value in data.items():
            # If a field isn't a var, skip it (e.g. `_task_queue`).
            try:
                var = cls_vars[field_name]
            except KeyError:
                continue

            if value is None:
                if var.field_attrs.get("required"):
                    raise ValidationError(f"{field_name} is a required field")
                else:
                    return_data[field_name] = value
                    continue

            if isinstance(var, MultiObjectVar):
                queryset = var.field_attrs["queryset"].filter(pk__in=value)
                if queryset.count() < len(value):
                    # Not all objects found
                    found_pks = set(queryset.values_list("pk", flat=True))
                    not_found_pks = set(value).difference(found_pks)
                    raise queryset.model.DoesNotExist(
                        f"Failed to find requested objects for var {field_name}: [{', '.join(not_found_pks)}]"
                    )
                return_data[field_name] = var.field_attrs["queryset"].filter(pk__in=value)

            elif isinstance(var, ObjectVar):
                if isinstance(value, dict):
                    return_data[field_name] = var.field_attrs["queryset"].get(**value)
                else:
                    return_data[field_name] = var.field_attrs["queryset"].get(pk=value)
            elif isinstance(var, FileVar):
                return_data[field_name] = cls.load_file(value)
            # IPAddressVar is a netaddr.IPAddress object
            elif isinstance(var, IPAddressVar):
                return_data[field_name] = netaddr.IPAddress(value)
            # IPAddressWithMaskVar, IPNetworkVar are netaddr.IPNetwork objects
            elif isinstance(var, (IPAddressWithMaskVar, IPNetworkVar)):
                return_data[field_name] = netaddr.IPNetwork(value)
            else:
                return_data[field_name] = value

        return return_data

    @classmethod
    def validate_data(cls, data, files=None):
        cls_vars = cls._get_vars()

        if not isinstance(data, dict):
            raise ValidationError("Job data needs to be a dict")

        for k in data:
            if k not in cls_vars:
                raise ValidationError({k: "Job data contained an unknown property"})

        # defer validation to the form object
        f = cls.as_form(data=cls.deserialize_data(data), files=files)
        if not f.is_valid():
            raise ValidationError(f.errors)

        return f.cleaned_data

    @classmethod
    def prepare_job_kwargs(cls, job_kwargs):
        """Process dict and return kwargs that exist as ScriptVariables on this job."""
        job_vars = cls._get_vars()
        return {k: v for k, v in job_kwargs.items() if k in job_vars}

    @staticmethod
    def load_file(pk):
        """Load a file proxy stored in the database by primary key.

        Args:
            pk (uuid): Primary key of the `FileProxy` to retrieve

        Returns:
            (FileProxy): A File-like object
        """
        fp = FileProxy.objects.get(pk=pk)
        return fp.file

    @staticmethod
    def save_file(uploaded_file):
        """
        Save an uploaded file to the database as a file proxy and return the
        primary key.

        Args:
            uploaded_file (file): File handle of file to save to database

        Returns:
            (uuid): The pk of the `FileProxy` object
        """
        fp = FileProxy.objects.create(name=uploaded_file.name, file=uploaded_file)
        return fp.pk

    def delete_files(self, *files_to_delete):
        """Given an unpacked list of primary keys for `FileProxy` objects, delete them.

        Args:
            files_to_delete (*args): List of primary keys to delete

        Returns:
            (int): number of objects deleted
        """
        files = FileProxy.objects.filter(pk__in=files_to_delete)
        num = 0
        for fp in files:
            fp.delete()  # Call delete() on each, so `FileAttachment` is reaped
            num += 1
        self.logger.debug("Deleted %d file proxies", num, extra={"grouping": "post_run"})
        return num

    # Convenience functions

    def load_yaml(self, filename):
        """
        Return data from a YAML file
        """
        file_path = os.path.join(os.path.dirname(self.file_path), filename)
        with open(file_path, "r") as datafile:
            data = yaml.safe_load(datafile)

        return data

    def load_json(self, filename):
        """
        Return data from a JSON file
        """
        file_path = os.path.join(os.path.dirname(self.file_path), filename)
        with open(file_path, "r") as datafile:
            data = json.load(datafile)

        return data

`Meta` ¶

Metaclass attributes - subclasses can define any or all of the following attributes:

name (str)
description (str)
hidden (bool)
field_order (list)
approval_required (bool)
soft_time_limit (int)
time_limit (int)
has_sensitive_variables (bool)
task_queues (list)

Source code in nautobot/extras/jobs.py

class Meta:
    """
    Metaclass attributes - subclasses can define any or all of the following attributes:

    - name (str)
    - description (str)
    - hidden (bool)
    - field_order (list)
    - approval_required (bool)
    - soft_time_limit (int)
    - time_limit (int)
    - has_sensitive_variables (bool)
    - task_queues (list)
    """

`after_return(status, retval, task_id, args, kwargs, einfo)` ¶

Handler called after the task returns.

Parameters status - Current task state. retval - Task return value/exception. task_id - Unique id of the task. args - Original arguments for the task that returned. kwargs - Original keyword arguments for the task that returned.

Keyword Arguments einfo - ExceptionInfo instance, containing the traceback (if any).

Returns:

Type	Description
`None`	The return value of this handler is ignored.

Source code in nautobot/extras/jobs.py

def after_return(self, status, retval, task_id, args, kwargs, einfo):
    """
    Handler called after the task returns.

    Parameters
        status - Current task state.
        retval - Task return value/exception.
        task_id - Unique id of the task.
        args - Original arguments for the task that returned.
        kwargs - Original keyword arguments for the task that returned.

    Keyword Arguments
        einfo - ExceptionInfo instance, containing the traceback (if any).

    Returns:
        (None): The return value of this handler is ignored.
    """

    # Cleanup FileProxy objects
    file_fields = list(self._get_file_vars())
    file_ids = [kwargs[f] for f in file_fields]
    if file_ids:
        self.delete_files(*file_ids)

    self.logger.info("Job completed", extra={"grouping": "post_run"})

    # TODO(gary): document this in job author docs
    # Super.after_return must be called for chords to function properly
    super().after_return(status, retval, task_id, args, kwargs, einfo=einfo)

`apply(args=None, kwargs=None, link=None, link_error=None, task_id=None, retries=None, throw=None, logfile=None, loglevel=None, headers=None, **options)` ¶

Fix celery's apply method to propagate options to the task result

Source code in nautobot/extras/jobs.py

def apply(
    self,
    args=None,
    kwargs=None,
    link=None,
    link_error=None,
    task_id=None,
    retries=None,
    throw=None,
    logfile=None,
    loglevel=None,
    headers=None,
    **options,
):
    """Fix celery's apply method to propagate options to the task result"""
    # trace imports Task, so need to import inline.
    from celery.app.trace import build_tracer

    app = self._get_app()
    args = args or ()
    kwargs = kwargs or {}
    task_id = task_id or uuid()
    retries = retries or 0
    if throw is None:
        throw = app.conf.task_eager_propagates

    # Make sure we get the task instance, not class.
    task = app._tasks[self.name]

    request = {
        "id": task_id,
        "retries": retries,
        "is_eager": True,
        "logfile": logfile,
        "loglevel": loglevel or 0,
        "hostname": gethostname(),
        "callbacks": maybe_list(link),
        "errbacks": maybe_list(link_error),
        "headers": headers,
        "ignore_result": options.get("ignore_result", False),
        "delivery_info": {
            "is_eager": True,
            "exchange": options.get("exchange"),
            "routing_key": options.get("routing_key"),
            "priority": options.get("priority"),
        },
        "properties": options,  # one line fix to overloaded method
    }
    if "stamped_headers" in options:
        request["stamped_headers"] = maybe_list(options["stamped_headers"])
        request["stamps"] = {header: maybe_list(options.get(header, [])) for header in request["stamped_headers"]}

    tb = None
    tracer = build_tracer(
        task.name,
        task,
        eager=True,
        propagate=throw,
        app=self._get_app(),
    )
    ret = tracer(task_id, args, kwargs, request)
    retval = ret.retval
    if isinstance(retval, ExceptionInfo):
        retval, tb = retval.exception, retval.traceback
        if isinstance(retval, ExceptionWithTraceback):
            retval = retval.exc
    if isinstance(retval, Retry) and retval.sig is not None:
        return retval.sig.apply(retries=retries + 1)
    state = states.SUCCESS if ret.info is None else ret.info.state
    return EagerResult(task_id, retval, state, traceback=tb)

`as_form(data=None, files=None, initial=None, approval_view=False)` `classmethod` ¶

Return a Django form suitable for populating the context data required to run this Job.

approval_view will disable all fields from modification and is used to display the form during a approval review workflow.

Source code in nautobot/extras/jobs.py

@classmethod
def as_form(cls, data=None, files=None, initial=None, approval_view=False):
    """
    Return a Django form suitable for populating the context data required to run this Job.

    `approval_view` will disable all fields from modification and is used to display the form
    during a approval review workflow.
    """

    form = cls.as_form_class()(data, files, initial=initial)

    try:
        job_model = JobModel.objects.get_for_class_path(cls.class_path)
        dryrun_default = job_model.dryrun_default if job_model.dryrun_default_override else cls.dryrun_default
        task_queues = job_model.task_queues if job_model.task_queues_override else cls.task_queues
    except JobModel.DoesNotExist:
        logger.error("No Job instance found in the database corresponding to %s", cls.class_path)
        dryrun_default = cls.dryrun_default
        task_queues = cls.task_queues

    # Update task queue choices
    form.fields["_task_queue"].choices = task_queues_as_choices(task_queues)

    if cls.supports_dryrun and (not initial or "dryrun" not in initial):
        # Set initial "dryrun" checkbox state based on the Meta parameter
        form.fields["dryrun"].initial = dryrun_default
    if not settings.DEBUG:
        form.fields["_profile"].widget = forms.HiddenInput()

    # https://github.com/PyCQA/pylint/issues/3484
    if cls.field_order:  # pylint: disable=using-constant-test
        form.order_fields(cls.field_order)

    if approval_view:
        # Set `disabled=True` on all fields
        for _, field in form.fields.items():
            field.disabled = True

    return form

`as_form_class()` `classmethod` ¶

Dynamically generate a Django form class corresponding to the variables in this Job.

In most cases you should use .as_form() instead of calling this method directly.

Source code in nautobot/extras/jobs.py

@classmethod
def as_form_class(cls):
    """
    Dynamically generate a Django form class corresponding to the variables in this Job.

    In most cases you should use `.as_form()` instead of calling this method directly.
    """
    fields = {name: var.as_field() for name, var in cls._get_vars().items()}
    return type("JobForm", (JobForm,), fields)

`before_start(task_id, args, kwargs)` ¶

Handler called before the task starts.

Parameters:

Name	Type	Description	Default
`task_id`	`str`	Unique id of the task to execute.	required
`args`	`Tuple`	Original arguments for the task to execute.	required
`kwargs`	`Dict`	Original keyword arguments for the task to execute.	required

Returns:

Type	Description
`None`	The return value of this handler is ignored.

Source code in nautobot/extras/jobs.py

def before_start(self, task_id, args, kwargs):
    """Handler called before the task starts.

    Arguments:
        task_id (str): Unique id of the task to execute.
        args (Tuple): Original arguments for the task to execute.
        kwargs (Dict): Original keyword arguments for the task to execute.

    Returns:
        (None): The return value of this handler is ignored.
    """
    self.clear_cache()

    try:
        self.job_result
    except ObjectDoesNotExist as err:
        raise RunJobTaskFailed(f"Unable to find associated job result for job {task_id}") from err

    try:
        self.job_model
    except ObjectDoesNotExist as err:
        raise RunJobTaskFailed(f"Unable to find associated job model for job {task_id}") from err

    if not self.job_model.enabled:
        self.logger.error(
            "Job %s is not enabled to be run!",
            self.job_model,
            extra={"object": self.job_model, "grouping": "initialization"},
        )
        raise RunJobTaskFailed(f"Job {self.job_model} is not enabled to be run!")

    soft_time_limit = self.job_model.soft_time_limit or settings.CELERY_TASK_SOFT_TIME_LIMIT
    time_limit = self.job_model.time_limit or settings.CELERY_TASK_TIME_LIMIT
    if time_limit <= soft_time_limit:
        self.logger.warning(
            "The hard time limit of %s seconds is less than "
            "or equal to the soft time limit of %s seconds. "
            "This job will fail silently after %s seconds.",
            time_limit,
            soft_time_limit,
            time_limit,
            extra={"grouping": "initialization"},
        )

    self.logger.info("Running job", extra={"grouping": "initialization"})

`class_path()` ¶

Unique identifier of a specific Job class, in the form ..

Examples:

my_script.MyScript - Local Job nautobot.core.jobs.MySystemJob - System Job my_plugin.jobs.MyPluginJob - App-provided Job git_repository.jobs.myjob.MyJob - GitRepository Job

Source code in nautobot/extras/jobs.py

@final
@classproperty
def class_path(cls) -> str:  # pylint: disable=no-self-argument
    """
    Unique identifier of a specific Job class, in the form <module_name>.<ClassName>.

    Examples:
    my_script.MyScript - Local Job
    nautobot.core.jobs.MySystemJob - System Job
    my_plugin.jobs.MyPluginJob - App-provided Job
    git_repository.jobs.myjob.MyJob - GitRepository Job
    """
    return f"{cls.__module__}.{cls.__name__}"

`class_path_dotted()` ¶

Dotted class_path, suitable for use in things like Python logger names.

Deprecated as of Nautobot 2.0: just use .class_path instead.

Source code in nautobot/extras/jobs.py

@final
@classproperty
def class_path_dotted(cls) -> str:  # pylint: disable=no-self-argument
    """
    Dotted class_path, suitable for use in things like Python logger names.

    Deprecated as of Nautobot 2.0: just use .class_path instead.
    """
    return cls.class_path

`class_path_js_escaped()` ¶

Escape various characters so that the class_path can be used as a jQuery selector.

Source code in nautobot/extras/jobs.py

@final
@classproperty
def class_path_js_escaped(cls) -> str:  # pylint: disable=no-self-argument
    """
    Escape various characters so that the class_path can be used as a jQuery selector.
    """
    return cls.class_path.replace(".", r"\.")

`clear_cache()` ¶

Clear all cached properties on this instance without accessing them. This is required because celery reuses task instances for multiple runs.

Source code in nautobot/extras/jobs.py

def clear_cache(self):
    """
    Clear all cached properties on this instance without accessing them. This is required because
    celery reuses task instances for multiple runs.
    """
    try:
        del self.celery_kwargs
    except AttributeError:
        pass
    try:
        del self.job_result
    except AttributeError:
        pass
    try:
        del self.job_model
    except AttributeError:
        pass

`delete_files(*files_to_delete)` ¶

Given an unpacked list of primary keys for FileProxy objects, delete them.

Parameters:

Name	Type	Description	Default
`files_to_delete`	`*args`	List of primary keys to delete	`()`

Returns:

Type	Description
`int`	number of objects deleted

Source code in nautobot/extras/jobs.py

def delete_files(self, *files_to_delete):
    """Given an unpacked list of primary keys for `FileProxy` objects, delete them.

    Args:
        files_to_delete (*args): List of primary keys to delete

    Returns:
        (int): number of objects deleted
    """
    files = FileProxy.objects.filter(pk__in=files_to_delete)
    num = 0
    for fp in files:
        fp.delete()  # Call delete() on each, so `FileAttachment` is reaped
        num += 1
    self.logger.debug("Deleted %d file proxies", num, extra={"grouping": "post_run"})
    return num

`deserialize_data(data)` `classmethod` ¶

Given data input for a job execution, deserialize it by resolving object references using defined variables.

This converts a list of pk's back into a QuerySet for MultiObjectVar instances and single pk values into model instances for ObjectVar.

Note that when resolving querysets or model instances by their PK, we do not catch DoesNotExist exceptions here, we leave it up the caller to handle those cases. The normal job execution code path would consider this a failure of the job execution, as described in nautobot.extras.jobs.run_job.

Source code in nautobot/extras/jobs.py

@classmethod
def deserialize_data(cls, data):
    """
    Given data input for a job execution, deserialize it by resolving object references using defined variables.

    This converts a list of pk's back into a QuerySet for MultiObjectVar instances and single pk values into
    model instances for ObjectVar.

    Note that when resolving querysets or model instances by their PK, we do not catch DoesNotExist
    exceptions here, we leave it up the caller to handle those cases. The normal job execution code
    path would consider this a failure of the job execution, as described in `nautobot.extras.jobs.run_job`.
    """
    cls_vars = cls._get_vars()
    return_data = {}

    if not isinstance(data, dict):
        raise TypeError("Data should be a dictionary.")

    for field_name, value in data.items():
        # If a field isn't a var, skip it (e.g. `_task_queue`).
        try:
            var = cls_vars[field_name]
        except KeyError:
            continue

        if value is None:
            if var.field_attrs.get("required"):
                raise ValidationError(f"{field_name} is a required field")
            else:
                return_data[field_name] = value
                continue

        if isinstance(var, MultiObjectVar):
            queryset = var.field_attrs["queryset"].filter(pk__in=value)
            if queryset.count() < len(value):
                # Not all objects found
                found_pks = set(queryset.values_list("pk", flat=True))
                not_found_pks = set(value).difference(found_pks)
                raise queryset.model.DoesNotExist(
                    f"Failed to find requested objects for var {field_name}: [{', '.join(not_found_pks)}]"
                )
            return_data[field_name] = var.field_attrs["queryset"].filter(pk__in=value)

        elif isinstance(var, ObjectVar):
            if isinstance(value, dict):
                return_data[field_name] = var.field_attrs["queryset"].get(**value)
            else:
                return_data[field_name] = var.field_attrs["queryset"].get(pk=value)
        elif isinstance(var, FileVar):
            return_data[field_name] = cls.load_file(value)
        # IPAddressVar is a netaddr.IPAddress object
        elif isinstance(var, IPAddressVar):
            return_data[field_name] = netaddr.IPAddress(value)
        # IPAddressWithMaskVar, IPNetworkVar are netaddr.IPNetwork objects
        elif isinstance(var, (IPAddressWithMaskVar, IPNetworkVar)):
            return_data[field_name] = netaddr.IPNetwork(value)
        else:
            return_data[field_name] = value

    return return_data

`load_file(pk)` `staticmethod` ¶

Load a file proxy stored in the database by primary key.

Parameters:

Name	Type	Description	Default
`pk`	`uuid`	Primary key of the `FileProxy` to retrieve	required

Returns:

Type	Description
`FileProxy`	A File-like object

Source code in nautobot/extras/jobs.py

@staticmethod
def load_file(pk):
    """Load a file proxy stored in the database by primary key.

    Args:
        pk (uuid): Primary key of the `FileProxy` to retrieve

    Returns:
        (FileProxy): A File-like object
    """
    fp = FileProxy.objects.get(pk=pk)
    return fp.file

`load_json(filename)` ¶

Return data from a JSON file

Source code in nautobot/extras/jobs.py

def load_json(self, filename):
    """
    Return data from a JSON file
    """
    file_path = os.path.join(os.path.dirname(self.file_path), filename)
    with open(file_path, "r") as datafile:
        data = json.load(datafile)

    return data

`load_yaml(filename)` ¶

Return data from a YAML file

Source code in nautobot/extras/jobs.py

def load_yaml(self, filename):
    """
    Return data from a YAML file
    """
    file_path = os.path.join(os.path.dirname(self.file_path), filename)
    with open(file_path, "r") as datafile:
        data = yaml.safe_load(datafile)

    return data

`on_bound(app)` `classmethod` ¶

Called when the task is bound to an app.

Note

This class method can be defined to do additional actions when the task class is bound to an app.

Source code in nautobot/extras/jobs.py

@classmethod
def on_bound(cls, app):
    """Called when the task is bound to an app.

    Note:
        This class method can be defined to do additional actions when
        the task class is bound to an app.
    """

`on_failure(exc, task_id, args, kwargs, einfo)` ¶

Error handler.

This is run by the worker when the task fails.

Parameters:

Name	Type	Description	Default
`exc`	`Exception`	The exception raised by the task.	required
`task_id`	`str`	Unique id of the failed task.	required
`args`	`Tuple`	Original arguments for the task that failed.	required
`kwargs`	`Dict`	Original keyword arguments for the task that failed.	required
`einfo`	`~billiard.einfo.ExceptionInfo`	Exception information.	required

Returns:

Type	Description
`None`	The return value of this handler is ignored.

Source code in nautobot/extras/jobs.py

def on_failure(self, exc, task_id, args, kwargs, einfo):
    """Error handler.

    This is run by the worker when the task fails.

    Arguments:
        exc (Exception): The exception raised by the task.
        task_id (str): Unique id of the failed task.
        args (Tuple): Original arguments for the task that failed.
        kwargs (Dict): Original keyword arguments for the task that failed.
        einfo (~billiard.einfo.ExceptionInfo): Exception information.

    Returns:
        (None): The return value of this handler is ignored.
    """

`on_retry(exc, task_id, args, kwargs, einfo)` ¶

Retry handler.

This is run by the worker when the task is to be retried.

Parameters:

Name	Type	Description	Default
`exc`	`Exception`	The exception sent to :meth:`retry`.	required
`task_id`	`str`	Unique id of the retried task.	required
`args`	`Tuple`	Original arguments for the retried task.	required
`kwargs`	`Dict`	Original keyword arguments for the retried task.	required
`einfo`	`~billiard.einfo.ExceptionInfo`	Exception information.	required

Returns:

Type	Description
`None`	The return value of this handler is ignored.

Source code in nautobot/extras/jobs.py

def on_retry(self, exc, task_id, args, kwargs, einfo):
    """Retry handler.

    This is run by the worker when the task is to be retried.

    Arguments:
        exc (Exception): The exception sent to :meth:`retry`.
        task_id (str): Unique id of the retried task.
        args (Tuple): Original arguments for the retried task.
        kwargs (Dict): Original keyword arguments for the retried task.
        einfo (~billiard.einfo.ExceptionInfo): Exception information.

    Returns:
        (None): The return value of this handler is ignored.
    """

`on_success(retval, task_id, args, kwargs)` ¶

Success handler.

Run by the worker if the task executes successfully.

Parameters:

Name	Type	Description	Default
`retval`	`Any`	The return value of the task.	required
`task_id`	`str`	Unique id of the executed task.	required
`args`	`Tuple`	Original arguments for the executed task.	required
`kwargs`	`Dict`	Original keyword arguments for the executed task.	required

Returns:

Type	Description
`None`	The return value of this handler is ignored.

Source code in nautobot/extras/jobs.py

def on_success(self, retval, task_id, args, kwargs):
    """Success handler.

    Run by the worker if the task executes successfully.

    Arguments:
        retval (Any): The return value of the task.
        task_id (str): Unique id of the executed task.
        args (Tuple): Original arguments for the executed task.
        kwargs (Dict): Original keyword arguments for the executed task.

    Returns:
        (None): The return value of this handler is ignored.
    """

`prepare_job_kwargs(job_kwargs)` `classmethod` ¶

Process dict and return kwargs that exist as ScriptVariables on this job.

Source code in nautobot/extras/jobs.py

@classmethod
def prepare_job_kwargs(cls, job_kwargs):
    """Process dict and return kwargs that exist as ScriptVariables on this job."""
    job_vars = cls._get_vars()
    return {k: v for k, v in job_kwargs.items() if k in job_vars}

`properties_dict()` ¶

Return all relevant classproperties as a dict.

Used for convenient rendering into job_edit.html via the json_script template tag.

Source code in nautobot/extras/jobs.py

@final
@classproperty
def properties_dict(cls) -> dict:  # pylint: disable=no-self-argument
    """
    Return all relevant classproperties as a dict.

    Used for convenient rendering into job_edit.html via the `json_script` template tag.
    """
    return {
        "name": cls.name,
        "grouping": cls.grouping,
        "description": cls.description,
        "approval_required": cls.approval_required,
        "hidden": cls.hidden,
        "soft_time_limit": cls.soft_time_limit,
        "time_limit": cls.time_limit,
        "has_sensitive_variables": cls.has_sensitive_variables,
        "task_queues": cls.task_queues,
    }

`run(*args, **kwargs)` ¶

Method invoked when this Job is run.

Source code in nautobot/extras/jobs.py

def run(self, *args, **kwargs):
    """
    Method invoked when this Job is run.
    """
    raise NotImplementedError("Jobs must define the run method.")

`save_file(uploaded_file)` `staticmethod` ¶

Save an uploaded file to the database as a file proxy and return the primary key.

Parameters:

Name	Type	Description	Default
`uploaded_file`	`file`	File handle of file to save to database	required

Returns:

Type	Description
`uuid`	The pk of the `FileProxy` object

Source code in nautobot/extras/jobs.py

@staticmethod
def save_file(uploaded_file):
    """
    Save an uploaded file to the database as a file proxy and return the
    primary key.

    Args:
        uploaded_file (file): File handle of file to save to database

    Returns:
        (uuid): The pk of the `FileProxy` object
    """
    fp = FileProxy.objects.create(name=uploaded_file.name, file=uploaded_file)
    return fp.pk

`serialize_data(data)` `staticmethod` ¶

This method parses input data (from JobForm usually) and returns a dict which is safe to serialize

Here we convert the QuerySet of a MultiObjectVar to a list of the pk's and the model instance of an ObjectVar into the pk value.

These are converted back during job execution.

Source code in nautobot/extras/jobs.py

@staticmethod
def serialize_data(data):
    """
    This method parses input data (from JobForm usually) and returns a dict which is safe to serialize

    Here we convert the QuerySet of a MultiObjectVar to a list of the pk's and the model instance
    of an ObjectVar into the pk value.

    These are converted back during job execution.
    """

    return_data = {}
    for field_name, value in data.items():
        # MultiObjectVar
        if isinstance(value, QuerySet):
            return_data[field_name] = list(value.values_list("pk", flat=True))
        # ObjectVar
        elif isinstance(value, Model):
            return_data[field_name] = value.pk
        # FileVar (Save each FileVar as a FileProxy)
        elif isinstance(value, InMemoryUploadedFile):
            return_data[field_name] = BaseJob.save_file(value)
        # IPAddressVar, IPAddressWithMaskVar, IPNetworkVar
        elif isinstance(value, netaddr.ip.BaseIP):
            return_data[field_name] = str(value)
        # Everything else...
        else:
            return_data[field_name] = value

    return return_data

`shadow_name(args, kwargs, options)` ¶

Override for custom task name in worker logs/monitoring.

Example

from celery.utils.imports import qualname

def shadow_name(task, args, kwargs, options): return qualname(args[0])

@app.task(shadow_name=shadow_name, serializer='pickle') def apply_function_async(fun, args, kwargs): return fun(args, **kwargs)

Parameters:

Name	Type	Description	Default
`args`	`Tuple`	Task positional arguments.	required
`kwargs`	`Dict`	Task keyword arguments.	required
`options`	`Dict`	Task execution options.	required

Source code in nautobot/extras/jobs.py

def shadow_name(self, args, kwargs, options):
    """Override for custom task name in worker logs/monitoring.

    Example:
            from celery.utils.imports import qualname

            def shadow_name(task, args, kwargs, options):
                return qualname(args[0])

            @app.task(shadow_name=shadow_name, serializer='pickle')
            def apply_function_async(fun, *args, **kwargs):
                return fun(*args, **kwargs)

    Arguments:
        args (Tuple): Task positional arguments.
        kwargs (Dict): Task keyword arguments.
        options (Dict): Task execution options.
    """

`nautobot.apps.jobs.BooleanVar` ¶

Bases: ScriptVariable

Boolean representation (true/false). Renders as a checkbox.

Source code in nautobot/extras/jobs.py

class BooleanVar(ScriptVariable):
    """
    Boolean representation (true/false). Renders as a checkbox.
    """

    form_field = forms.BooleanField

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        # Boolean fields cannot be required
        self.field_attrs["required"] = False

`nautobot.apps.jobs.ChoiceVar` ¶

Bases: ScriptVariable

Select one of several predefined static choices, passed as a list of two-tuples. Example:

color = ChoiceVar(
    choices=(
        ('#ff0000', 'Red'),
        ('#00ff00', 'Green'),
        ('#0000ff', 'Blue')
    )
)

Source code in nautobot/extras/jobs.py

class ChoiceVar(ScriptVariable):
    """
    Select one of several predefined static choices, passed as a list of two-tuples. Example:

        color = ChoiceVar(
            choices=(
                ('#ff0000', 'Red'),
                ('#00ff00', 'Green'),
                ('#0000ff', 'Blue')
            )
        )
    """

    form_field = forms.ChoiceField

    def __init__(self, choices, *args, **kwargs):
        super().__init__(*args, **kwargs)

        # Set field choices
        self.field_attrs["choices"] = choices

`nautobot.apps.jobs.DatabaseFileField` ¶

Bases: forms.FileField

Specialized FileField for use with DatabaseFileStorage storage backend.

Source code in nautobot/extras/jobs.py

class DatabaseFileField(forms.FileField):
    """Specialized `FileField` for use with `DatabaseFileStorage` storage backend."""

    widget = DBClearableFileInput

`nautobot.apps.jobs.DryRunVar` ¶

Bases: BooleanVar

Special boolean variable that bypasses approval requirements if this is set to True on job execution.

Source code in nautobot/extras/jobs.py

class DryRunVar(BooleanVar):
    """
    Special boolean variable that bypasses approval requirements if this is set to True on job execution.
    """

    description = "Check to run job in dryrun mode."

    def __init__(self, *args, **kwargs):
        # Default must be false unless overridden through `dryrun_default` meta attribute
        kwargs["default"] = False

        # Default description if one was not provided
        kwargs.setdefault("description", self.description)

        super().__init__(*args, **kwargs)

`nautobot.apps.jobs.FileVar` ¶

Bases: ScriptVariable

An uploaded file.

Source code in nautobot/extras/jobs.py

class FileVar(ScriptVariable):
    """
    An uploaded file.
    """

    form_field = DatabaseFileField

`nautobot.apps.jobs.GitRepositoryDryRun` ¶

Bases: Job

System Job to perform a dry run on a Git repository.

Source code in nautobot/core/jobs/__init__.py

class GitRepositoryDryRun(Job):
    """System Job to perform a dry run on a Git repository."""

    repository = ObjectVar(
        description="Git Repository to dry-run",
        label="Git Repository",
        model=GitRepository,
    )

    class Meta:
        name = "Git Repository: Dry-Run"
        has_sensitive_variables = False

    def run(self, repository):
        job_result = self.job_result
        self.logger.info(f'Performing a Dry Run on Git repository "{repository.name}"...')

        try:
            git_repository_dry_run(repository, logger=self.logger)
        finally:
            self.logger.info(f"Repository dry run completed in {job_result.duration}")

`nautobot.apps.jobs.GitRepositorySync` ¶

Bases: Job

System job to clone and/or pull a Git repository, then invoke refresh_datasource_content().

Source code in nautobot/core/jobs/__init__.py

class GitRepositorySync(Job):
    """
    System job to clone and/or pull a Git repository, then invoke `refresh_datasource_content()`.
    """

    repository = ObjectVar(
        description="Git Repository to pull and refresh",
        label="Git Repository",
        model=GitRepository,
    )

    class Meta:
        name = "Git Repository: Sync"
        has_sensitive_variables = False

    def run(self, repository):
        job_result = self.job_result
        user = job_result.user

        self.logger.info(f'Creating/refreshing local copy of Git repository "{repository.name}"...')

        try:
            ensure_git_repository(repository, logger=self.logger)
            refresh_datasource_content("extras.gitrepository", repository, user, job_result, delete=False)
            # Given that the above succeeded, tell all workers (including ourself) to call ensure_git_repository()
            app.control.broadcast("refresh_git_repository", repository_pk=repository.pk, head=repository.current_head)
        finally:
            self.logger.info(f"Repository synchronization completed in {job_result.duration}")

`nautobot.apps.jobs.IPAddressVar` ¶

Bases: ScriptVariable

An IPv4 or IPv6 address without a mask.

Source code in nautobot/extras/jobs.py

class IPAddressVar(ScriptVariable):
    """
    An IPv4 or IPv6 address without a mask.
    """

    form_field = IPAddressFormField

`nautobot.apps.jobs.IPAddressWithMaskVar` ¶

Bases: ScriptVariable

An IPv4 or IPv6 address with a mask.

Source code in nautobot/extras/jobs.py

class IPAddressWithMaskVar(ScriptVariable):
    """
    An IPv4 or IPv6 address with a mask.
    """

    form_field = IPNetworkFormField

`nautobot.apps.jobs.IPNetworkVar` ¶

Bases: ScriptVariable

An IPv4 or IPv6 prefix.

Source code in nautobot/extras/jobs.py

class IPNetworkVar(ScriptVariable):
    """
    An IPv4 or IPv6 prefix.
    """

    form_field = IPNetworkFormField

    def __init__(self, min_prefix_length=None, max_prefix_length=None, *args, **kwargs):
        super().__init__(*args, **kwargs)

        # Set prefix validator and optional minimum/maximum prefix lengths
        self.field_attrs["validators"] = [prefix_validator]
        if min_prefix_length is not None:
            self.field_attrs["validators"].append(MinPrefixLengthValidator(min_prefix_length))
        if max_prefix_length is not None:
            self.field_attrs["validators"].append(MaxPrefixLengthValidator(max_prefix_length))

`nautobot.apps.jobs.IntegerVar` ¶

Bases: ScriptVariable

Integer representation. Can enforce minimum/maximum values.

Source code in nautobot/extras/jobs.py

class IntegerVar(ScriptVariable):
    """
    Integer representation. Can enforce minimum/maximum values.
    """

    form_field = forms.IntegerField

    def __init__(self, min_value=None, max_value=None, *args, **kwargs):
        super().__init__(*args, **kwargs)

        # Optional minimum/maximum values
        if min_value:
            self.field_attrs["min_value"] = min_value
        if max_value:
            self.field_attrs["max_value"] = max_value

`nautobot.apps.jobs.Job` ¶

Bases: BaseJob

Classes which inherit from this model will appear in the list of available jobs.

Source code in nautobot/extras/jobs.py

class Job(BaseJob):
    """
    Classes which inherit from this model will appear in the list of available jobs.
    """

`nautobot.apps.jobs.JobButtonReceiver` ¶

Bases: Job

Base class for job button receivers. Job button receivers are jobs that are initiated from UI buttons and are not intended to be run from the job form UI or API like standard jobs.

Source code in nautobot/extras/jobs.py

class JobButtonReceiver(Job):
    """
    Base class for job button receivers. Job button receivers are jobs that are initiated
    from UI buttons and are not intended to be run from the job form UI or API like standard jobs.
    """

    object_pk = StringVar()
    object_model_name = StringVar()

    def run(self, object_pk, object_model_name):
        """JobButtonReceiver subclasses generally shouldn't need to override this method."""
        model = get_model_from_name(object_model_name)
        obj = model.objects.get(pk=object_pk)

        self.receive_job_button(obj=obj)

    def receive_job_button(self, obj):
        """
        Method to be implemented by concrete JobButtonReceiver subclasses.

        :param obj: an instance of the object
        """
        raise NotImplementedError

`receive_job_button(obj)` ¶

Method to be implemented by concrete JobButtonReceiver subclasses.

:param obj: an instance of the object

Source code in nautobot/extras/jobs.py

def receive_job_button(self, obj):
    """
    Method to be implemented by concrete JobButtonReceiver subclasses.

    :param obj: an instance of the object
    """
    raise NotImplementedError

`run(object_pk, object_model_name)` ¶

JobButtonReceiver subclasses generally shouldn't need to override this method.

Source code in nautobot/extras/jobs.py

def run(self, object_pk, object_model_name):
    """JobButtonReceiver subclasses generally shouldn't need to override this method."""
    model = get_model_from_name(object_model_name)
    obj = model.objects.get(pk=object_pk)

    self.receive_job_button(obj=obj)

`nautobot.apps.jobs.JobHookReceiver` ¶

Bases: Job

Base class for job hook receivers. Job hook receivers are jobs that are initiated from object changes and are not intended to be run from the UI or API like standard jobs.

Source code in nautobot/extras/jobs.py

class JobHookReceiver(Job):
    """
    Base class for job hook receivers. Job hook receivers are jobs that are initiated
    from object changes and are not intended to be run from the UI or API like standard jobs.
    """

    object_change = ObjectVar(model=ObjectChange)

    def run(self, object_change):
        """JobHookReceiver subclasses generally shouldn't need to override this method."""
        self.receive_job_hook(
            change=object_change,
            action=object_change.action,
            changed_object=object_change.changed_object,
        )

    def receive_job_hook(self, change, action, changed_object):
        """
        Method to be implemented by concrete JobHookReceiver subclasses.

        :param change: an instance of `nautobot.extras.models.ObjectChange`
        :param action: a string with the action performed on the changed object ("create", "update" or "delete")
        :param changed_object: an instance of the object that was changed, or `None` if the object has been deleted
        """
        raise NotImplementedError

`receive_job_hook(change, action, changed_object)` ¶

Method to be implemented by concrete JobHookReceiver subclasses.

:param change: an instance of nautobot.extras.models.ObjectChange :param action: a string with the action performed on the changed object ("create", "update" or "delete") :param changed_object: an instance of the object that was changed, or None if the object has been deleted

Source code in nautobot/extras/jobs.py

def receive_job_hook(self, change, action, changed_object):
    """
    Method to be implemented by concrete JobHookReceiver subclasses.

    :param change: an instance of `nautobot.extras.models.ObjectChange`
    :param action: a string with the action performed on the changed object ("create", "update" or "delete")
    :param changed_object: an instance of the object that was changed, or `None` if the object has been deleted
    """
    raise NotImplementedError

`run(object_change)` ¶

JobHookReceiver subclasses generally shouldn't need to override this method.

Source code in nautobot/extras/jobs.py

def run(self, object_change):
    """JobHookReceiver subclasses generally shouldn't need to override this method."""
    self.receive_job_hook(
        change=object_change,
        action=object_change.action,
        changed_object=object_change.changed_object,
    )

`nautobot.apps.jobs.MultiChoiceVar` ¶

Bases: ChoiceVar

Like ChoiceVar, but allows for the selection of multiple choices.

Source code in nautobot/extras/jobs.py

class MultiChoiceVar(ChoiceVar):
    """
    Like ChoiceVar, but allows for the selection of multiple choices.
    """

    form_field = forms.MultipleChoiceField

`nautobot.apps.jobs.MultiObjectVar` ¶

Bases: ObjectVar

Like ObjectVar, but can represent one or more objects.

Source code in nautobot/extras/jobs.py

class MultiObjectVar(ObjectVar):
    """
    Like ObjectVar, but can represent one or more objects.
    """

    form_field = DynamicModelMultipleChoiceField

`nautobot.apps.jobs.NautobotKombuJSONEncoder` ¶

Bases: JSONEncoder

Custom json encoder based on restframework's JSONEncoder that serializes objects that implement the nautobot_serialize() method via the __nautobot_type__ interface. This is useful in passing special objects to and from Celery tasks.

This pattern should generally be avoided by passing pointers to persisted objects to the Celery tasks and retrieving them from within the task execution. While this is always possible for model instances (which covers 99% of use cases), for rare instances where it does not, and the actual object must be passed, this pattern allows for encoding and decoding of such objects.

It requires a conforming class to implement the instance method nautobot_serialize() which returns a json serializable dictionary of the object representation. The class must also implement the nautobot_deserialize() class method which takes the dictionary representation and returns an actual instance of the class.

Source code in nautobot/core/celery/encoders.py

class NautobotKombuJSONEncoder(JSONEncoder):
    """
    Custom json encoder based on restframework's JSONEncoder that serializes objects that implement
    the `nautobot_serialize()` method via the `__nautobot_type__` interface. This is useful
    in passing special objects to and from Celery tasks.

    This pattern should generally be avoided by passing pointers to persisted objects to the
    Celery tasks and retrieving them from within the task execution. While this is always possible
    for model instances (which covers 99% of use cases), for rare instances where it does not,
    and the actual object must be passed, this pattern allows for encoding and decoding
    of such objects.

    It requires a conforming class to implement the instance method `nautobot_serialize()` which
    returns a json serializable dictionary of the object representation. The class must also implement
    the `nautobot_deserialize()` class method which takes the dictionary representation and returns
    an actual instance of the class.
    """

    def default(self, obj):
        # Import here to avoid django.core.exceptions.ImproperlyConfigured Error.
        # Core App is not set up yet if we import this at the top of the file.
        from nautobot.core.models import BaseModel
        from nautobot.core.models.managers import TagsManager

        if isinstance(obj, BaseModel):
            cls = obj.__class__
            module = cls.__module__
            qual_name = ".".join([module, cls.__qualname__])  # fully qualified dotted import path
            logger.debug("Performing nautobot serialization on %s for type %s", obj, qual_name)
            data = {
                "id": obj.id,
                "__nautobot_type__": qual_name,
                # TODO: change to natural key to provide additional context if object is deleted from the db
                "display": getattr(obj, "display", str(obj)),
            }
            return data

        elif isinstance(obj, set):
            # Convert a set to a list for passing to and from a task
            return list(obj)
        elif isinstance(obj, TagsManager):
            obj = obj.values_list("id", flat=True)
            return obj
        elif isinstance(obj, Exception):
            # JobResult.result uses NautobotKombuJSONEncoder as an encoder and expects a JSONSerializable object,
            # although an exception, such as a RuntimeException, can be supplied as the obj.
            return f"{obj.__class__.__name__}: {obj}"
        else:
            return super().default(obj)

`nautobot.apps.jobs.ObjectVar` ¶

Bases: ScriptVariable

A single object within Nautobot.

:param model: The Nautobot model being referenced :param display_field: The attribute of the returned object to display in the selection list (default: 'name') :param query_params: A dictionary of additional query parameters to attach when making REST API requests (optional) :param null_option: The label to use as a "null" selection option (optional)

Source code in nautobot/extras/jobs.py

class ObjectVar(ScriptVariable):
    """
    A single object within Nautobot.

    :param model: The Nautobot model being referenced
    :param display_field: The attribute of the returned object to display in the selection list (default: 'name')
    :param query_params: A dictionary of additional query parameters to attach when making REST API requests (optional)
    :param null_option: The label to use as a "null" selection option (optional)
    """

    form_field = DynamicModelChoiceField

    def __init__(
        self,
        model=None,
        queryset=None,
        display_field="display",
        query_params=None,
        null_option=None,
        *args,
        **kwargs,
    ):
        super().__init__(*args, **kwargs)

        # Set the form field's queryset. Support backward compatibility for the "queryset" argument for now.
        if model is not None:
            self.field_attrs["queryset"] = model.objects.all()
        elif queryset is not None:
            warnings.warn(
                f'{self}: Specifying a queryset for ObjectVar is no longer supported. Please use "model" instead.'
            )
            self.field_attrs["queryset"] = queryset
        else:
            raise TypeError("ObjectVar must specify a model")

        self.field_attrs.update(
            {
                "display_field": display_field,
                "query_params": query_params,
                "null_option": null_option,
            }
        )

`nautobot.apps.jobs.RunJobTaskFailed` ¶

Bases: Exception

Celery task failed for some reason.

Source code in nautobot/extras/jobs.py

class RunJobTaskFailed(Exception):
    """Celery task failed for some reason."""

`nautobot.apps.jobs.ScriptVariable` ¶

Base model for script variables

Source code in nautobot/extras/jobs.py

class ScriptVariable:
    """
    Base model for script variables
    """

    form_field = forms.CharField

    def __init__(self, label="", description="", default=None, required=True, widget=None):
        # Initialize field attributes
        if not hasattr(self, "field_attrs"):
            self.field_attrs = {}
        if label:
            self.field_attrs["label"] = label
        if description:
            self.field_attrs["help_text"] = description
        if default is not None:
            self.field_attrs["initial"] = default
        if widget:
            self.field_attrs["widget"] = widget
        self.field_attrs["required"] = required

    def as_field(self):
        """
        Render the variable as a Django form field.
        """
        form_field = self.form_field(**self.field_attrs)
        if not isinstance(form_field.widget, forms.CheckboxInput):
            if form_field.widget.attrs and "class" in form_field.widget.attrs.keys():
                form_field.widget.attrs["class"] += " form-control"
            else:
                form_field.widget.attrs["class"] = "form-control"

        return form_field

`as_field()` ¶

Render the variable as a Django form field.

Source code in nautobot/extras/jobs.py

def as_field(self):
    """
    Render the variable as a Django form field.
    """
    form_field = self.form_field(**self.field_attrs)
    if not isinstance(form_field.widget, forms.CheckboxInput):
        if form_field.widget.attrs and "class" in form_field.widget.attrs.keys():
            form_field.widget.attrs["class"] += " form-control"
        else:
            form_field.widget.attrs["class"] = "form-control"

    return form_field

`nautobot.apps.jobs.StringVar` ¶

Bases: ScriptVariable

Character string representation. Can enforce minimum/maximum length and/or regex validation.

Source code in nautobot/extras/jobs.py

class StringVar(ScriptVariable):
    """
    Character string representation. Can enforce minimum/maximum length and/or regex validation.
    """

    def __init__(self, min_length=None, max_length=None, regex=None, *args, **kwargs):
        super().__init__(*args, **kwargs)

        # Optional minimum/maximum lengths
        if min_length:
            self.field_attrs["min_length"] = min_length
        if max_length:
            self.field_attrs["max_length"] = max_length

        # Optional regular expression validation
        if regex:
            self.field_attrs["validators"] = [
                RegexValidator(
                    regex=regex,
                    message=f"Invalid value. Must match regex: {regex}",
                    code="invalid",
                )
            ]

`nautobot.apps.jobs.TextVar` ¶

Bases: ScriptVariable

Free-form text data. Renders as a . <details class="quote"> <summary>Source code in <code>nautobot/extras/jobs.py</code></summary> <div class="highlight"><table class="highlighttable"><tr><td class="linenos"><div class="linenodiv"><pre><a href="#__codelineno-0-899">899</a> <a href="#__codelineno-0-900">900</a> <a href="#__codelineno-0-901">901</a> <a href="#__codelineno-0-902">902</a> <a href="#__codelineno-0-903">903</a> <a href="#__codelineno-0-904">904</a> <a href="#__codelineno-0-905">905</a> <a href="#__codelineno-0-906">906</a> <a href="#__codelineno-0-907">907</a> <a href="#__codelineno-0-908">908</a> <a href="#__codelineno-0-909">909</a></pre></div></td><td class="code"><div><pre><code><a id="__codelineno-0-899" name="__codelineno-0-899"></a>class TextVar(ScriptVariable): <a id="__codelineno-0-900" name="__codelineno-0-900"></a> """ <a id="__codelineno-0-901" name="__codelineno-0-901"></a> Free-form text data. Renders as a <textarea>. <a id="__codelineno-0-902" name="__codelineno-0-902"></a> """ <a id="__codelineno-0-903" name="__codelineno-0-903"></a> <a id="__codelineno-0-904" name="__codelineno-0-904"></a> form_field = forms.CharField <a id="__codelineno-0-905" name="__codelineno-0-905"></a> <a id="__codelineno-0-906" name="__codelineno-0-906"></a> def __init__(self, *args, **kwargs): <a id="__codelineno-0-907" name="__codelineno-0-907"></a> super().__init__(*args, **kwargs) <a id="__codelineno-0-908" name="__codelineno-0-908"></a> <a id="__codelineno-0-909" name="__codelineno-0-909"></a> self.field_attrs["widget"] = forms.Textarea </code></pre></div></td></tr></table></div> </details> <div class="doc doc-children"> </div> </div> </div> <div class="doc doc-object doc-function"> <h2 id="nautobot.apps.jobs.enqueue_job_hooks" class="doc doc-heading"> <code class="highlight language-python">nautobot.apps.jobs.enqueue_job_hooks(object_change)</code> <a href="#nautobot.apps.jobs.enqueue_job_hooks" class="headerlink" title="Permanent link">¶</a></h2> <div class="doc doc-contents "> Find job hook(s) assigned to this changed object type + action and enqueue them to be processed <details class="quote"> <summary>Source code in <code>nautobot/extras/jobs.py</code></summary> <div class="highlight"><table class="highlighttable"><tr><td class="linenos"><div class="linenodiv"><pre><a href="#__codelineno-0-1171">1171</a> <a href="#__codelineno-0-1172">1172</a> <a href="#__codelineno-0-1173">1173</a> <a href="#__codelineno-0-1174">1174</a> <a href="#__codelineno-0-1175">1175</a> <a href="#__codelineno-0-1176">1176</a> <a href="#__codelineno-0-1177">1177</a> <a href="#__codelineno-0-1178">1178</a> <a href="#__codelineno-0-1179">1179</a> <a href="#__codelineno-0-1180">1180</a> <a href="#__codelineno-0-1181">1181</a> <a href="#__codelineno-0-1182">1182</a> <a href="#__codelineno-0-1183">1183</a> <a href="#__codelineno-0-1184">1184</a> <a href="#__codelineno-0-1185">1185</a> <a href="#__codelineno-0-1186">1186</a> <a href="#__codelineno-0-1187">1187</a> <a href="#__codelineno-0-1188">1188</a> <a href="#__codelineno-0-1189">1189</a> <a href="#__codelineno-0-1190">1190</a> <a href="#__codelineno-0-1191">1191</a> <a href="#__codelineno-0-1192">1192</a> <a href="#__codelineno-0-1193">1193</a> <a href="#__codelineno-0-1194">1194</a> <a href="#__codelineno-0-1195">1195</a> <a href="#__codelineno-0-1196">1196</a> <a href="#__codelineno-0-1197">1197</a></pre></div></td><td class="code"><div><pre><code><a id="__codelineno-0-1171" name="__codelineno-0-1171"></a>def enqueue_job_hooks(object_change): <a id="__codelineno-0-1172" name="__codelineno-0-1172"></a> """ <a id="__codelineno-0-1173" name="__codelineno-0-1173"></a> Find job hook(s) assigned to this changed object type + action and enqueue them <a id="__codelineno-0-1174" name="__codelineno-0-1174"></a> to be processed <a id="__codelineno-0-1175" name="__codelineno-0-1175"></a> """ <a id="__codelineno-0-1176" name="__codelineno-0-1176"></a> <a id="__codelineno-0-1177" name="__codelineno-0-1177"></a> # Job hooks cannot trigger other job hooks <a id="__codelineno-0-1178" name="__codelineno-0-1178"></a> if object_change.change_context == ObjectChangeEventContextChoices.CONTEXT_JOB_HOOK: <a id="__codelineno-0-1179" name="__codelineno-0-1179"></a> return <a id="__codelineno-0-1180" name="__codelineno-0-1180"></a> <a id="__codelineno-0-1181" name="__codelineno-0-1181"></a> # Determine whether this type of object supports job hooks <a id="__codelineno-0-1182" name="__codelineno-0-1182"></a> content_type = object_change.changed_object_type <a id="__codelineno-0-1183" name="__codelineno-0-1183"></a> if content_type not in ChangeLoggedModelsQuery().as_queryset(): <a id="__codelineno-0-1184" name="__codelineno-0-1184"></a> return <a id="__codelineno-0-1185" name="__codelineno-0-1185"></a> <a id="__codelineno-0-1186" name="__codelineno-0-1186"></a> # Retrieve any applicable job hooks <a id="__codelineno-0-1187" name="__codelineno-0-1187"></a> action_flag = { <a id="__codelineno-0-1188" name="__codelineno-0-1188"></a> ObjectChangeActionChoices.ACTION_CREATE: "type_create", <a id="__codelineno-0-1189" name="__codelineno-0-1189"></a> ObjectChangeActionChoices.ACTION_UPDATE: "type_update", <a id="__codelineno-0-1190" name="__codelineno-0-1190"></a> ObjectChangeActionChoices.ACTION_DELETE: "type_delete", <a id="__codelineno-0-1191" name="__codelineno-0-1191"></a> }[object_change.action] <a id="__codelineno-0-1192" name="__codelineno-0-1192"></a> job_hooks = JobHook.objects.filter(content_types=content_type, enabled=True, **{action_flag: True}) <a id="__codelineno-0-1193" name="__codelineno-0-1193"></a> <a id="__codelineno-0-1194" name="__codelineno-0-1194"></a> # Enqueue the jobs related to the job_hooks <a id="__codelineno-0-1195" name="__codelineno-0-1195"></a> for job_hook in job_hooks: <a id="__codelineno-0-1196" name="__codelineno-0-1196"></a> job_model = job_hook.job <a id="__codelineno-0-1197" name="__codelineno-0-1197"></a> JobResult.enqueue_job(job_model, object_change.user, object_change=object_change.pk) </code></pre></div></td></tr></table></div> </details> </div> </div> <div class="doc doc-object doc-function"> <h2 id="nautobot.apps.jobs.get_job" class="doc doc-heading"> <code class="highlight language-python">nautobot.apps.jobs.get_job(class_path)</code> <a href="#nautobot.apps.jobs.get_job" class="headerlink" title="Permanent link">¶</a></h2> <div class="doc doc-contents "> Retrieve a specific job class by its class_path (<module_name>.<JobClassName>). May return None if the job isn't properly registered with Celery at this time. <details class="quote"> <summary>Source code in <code>nautobot/extras/jobs.py</code></summary> <div class="highlight"><table class="highlighttable"><tr><td class="linenos"><div class="linenodiv"><pre><a href="#__codelineno-0-1159">1159</a> <a href="#__codelineno-0-1160">1160</a> <a href="#__codelineno-0-1161">1161</a> <a href="#__codelineno-0-1162">1162</a> <a href="#__codelineno-0-1163">1163</a> <a href="#__codelineno-0-1164">1164</a> <a href="#__codelineno-0-1165">1165</a> <a href="#__codelineno-0-1166">1166</a> <a href="#__codelineno-0-1167">1167</a> <a href="#__codelineno-0-1168">1168</a></pre></div></td><td class="code"><div><pre><code><a id="__codelineno-0-1159" name="__codelineno-0-1159"></a>def get_job(class_path): <a id="__codelineno-0-1160" name="__codelineno-0-1160"></a> """ <a id="__codelineno-0-1161" name="__codelineno-0-1161"></a> Retrieve a specific job class by its class_path (<module_name>.<JobClassName>). <a id="__codelineno-0-1162" name="__codelineno-0-1162"></a> <a id="__codelineno-0-1163" name="__codelineno-0-1163"></a> May return None if the job isn't properly registered with Celery at this time. <a id="__codelineno-0-1164" name="__codelineno-0-1164"></a> """ <a id="__codelineno-0-1165" name="__codelineno-0-1165"></a> try: <a id="__codelineno-0-1166" name="__codelineno-0-1166"></a> return celery_app.tasks[class_path].__class__ <a id="__codelineno-0-1167" name="__codelineno-0-1167"></a> except NotRegistered: <a id="__codelineno-0-1168" name="__codelineno-0-1168"></a> return None </code></pre></div></td></tr></table></div> </details> </div> </div> <div class="doc doc-object doc-function"> <h2 id="nautobot.apps.jobs.is_job" class="doc doc-heading"> <code class="highlight language-python">nautobot.apps.jobs.is_job(obj)</code> <a href="#nautobot.apps.jobs.is_job" class="headerlink" title="Permanent link">¶</a></h2> <div class="doc doc-contents "> Returns True if the given object is a Job subclass. <details class="quote"> <summary>Source code in <code>nautobot/extras/jobs.py</code></summary> <div class="highlight"><table class="highlighttable"><tr><td class="linenos"><div class="linenodiv"><pre><a href="#__codelineno-0-1142">1142</a> <a href="#__codelineno-0-1143">1143</a> <a href="#__codelineno-0-1144">1144</a> <a href="#__codelineno-0-1145">1145</a> <a href="#__codelineno-0-1146">1146</a> <a href="#__codelineno-0-1147">1147</a> <a href="#__codelineno-0-1148">1148</a> <a href="#__codelineno-0-1149">1149</a></pre></div></td><td class="code"><div><pre><code><a id="__codelineno-0-1142" name="__codelineno-0-1142"></a>def is_job(obj): <a id="__codelineno-0-1143" name="__codelineno-0-1143"></a> """ <a id="__codelineno-0-1144" name="__codelineno-0-1144"></a> Returns True if the given object is a Job subclass. <a id="__codelineno-0-1145" name="__codelineno-0-1145"></a> """ <a id="__codelineno-0-1146" name="__codelineno-0-1146"></a> try: <a id="__codelineno-0-1147" name="__codelineno-0-1147"></a> return issubclass(obj, Job) and obj not in [Job, JobHookReceiver, JobButtonReceiver] <a id="__codelineno-0-1148" name="__codelineno-0-1148"></a> except TypeError: <a id="__codelineno-0-1149" name="__codelineno-0-1149"></a> return False </code></pre></div></td></tr></table></div> </details> </div> </div> <div class="doc doc-object doc-function"> <h2 id="nautobot.apps.jobs.is_variable" class="doc doc-heading"> <code class="highlight language-python">nautobot.apps.jobs.is_variable(obj)</code> <a href="#nautobot.apps.jobs.is_variable" class="headerlink" title="Permanent link">¶</a></h2> <div class="doc doc-contents "> Returns True if the object is a ScriptVariable instance. <details class="quote"> <summary>Source code in <code>nautobot/extras/jobs.py</code></summary> <div class="highlight"><table class="highlighttable"><tr><td class="linenos"><div class="linenodiv"><pre><a href="#__codelineno-0-1152">1152</a> <a href="#__codelineno-0-1153">1153</a> <a href="#__codelineno-0-1154">1154</a> <a href="#__codelineno-0-1155">1155</a> <a href="#__codelineno-0-1156">1156</a></pre></div></td><td class="code"><div><pre><code><a id="__codelineno-0-1152" name="__codelineno-0-1152"></a>def is_variable(obj): <a id="__codelineno-0-1153" name="__codelineno-0-1153"></a> """ <a id="__codelineno-0-1154" name="__codelineno-0-1154"></a> Returns True if the object is a ScriptVariable instance. <a id="__codelineno-0-1155" name="__codelineno-0-1155"></a> """ <a id="__codelineno-0-1156" name="__codelineno-0-1156"></a> return isinstance(obj, ScriptVariable) </code></pre></div></td></tr></table></div> </details> </div> </div> <div class="doc doc-object doc-function"> <h2 id="nautobot.apps.jobs.register_jobs" class="doc doc-heading"> <code class="highlight language-python">nautobot.apps.jobs.register_jobs(*jobs)</code> <a href="#nautobot.apps.jobs.register_jobs" class="headerlink" title="Permanent link">¶</a></h2> <div class="doc doc-contents "> Helper method to register jobs with Celery <details class="quote"> <summary>Source code in <code>nautobot/core/celery/__init__.py</code></summary> <div class="highlight"><table class="highlighttable"><tr><td class="linenos"><div class="linenodiv"><pre><a href="#__codelineno-0-225">225</a> <a href="#__codelineno-0-226">226</a> <a href="#__codelineno-0-227">227</a> <a href="#__codelineno-0-228">228</a> <a href="#__codelineno-0-229">229</a> <a href="#__codelineno-0-230">230</a></pre></div></td><td class="code"><div><pre><code><a id="__codelineno-0-225" name="__codelineno-0-225"></a>def register_jobs(*jobs): <a id="__codelineno-0-226" name="__codelineno-0-226"></a> """Helper method to register jobs with Celery""" <a id="__codelineno-0-227" name="__codelineno-0-227"></a> for job in jobs: <a id="__codelineno-0-228" name="__codelineno-0-228"></a> # TODO: should we only register a job if it corresponds to a Job database record? <a id="__codelineno-0-229" name="__codelineno-0-229"></a> logger.debug("Registering job %s.%s", job.__module__, job.__name__) <a id="__codelineno-0-230" name="__codelineno-0-230"></a> app.register_task(job) </code></pre></div></td></tr></table></div> </details> </div> </div> </div> </div> </div> </article> </div> </div> </main> <footer class="md-footer"> <nav class="md-footer__inner md-grid" aria-label="Footer" > <a href="graphql.html" class="md-footer__link md-footer__link--prev" aria-label="Previous: nautobot.apps.graphql" rel="prev"> <div class="md-footer__button md-icon"> <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24"><path d="M20 11v2H8l5.5 5.5-1.42 1.42L4.16 12l7.92-7.92L13.5 5.5 8 11h12Z"/></svg> </div> <div class="md-footer__title"> Previous <div class="md-ellipsis"> nautobot.apps.graphql </div> </div> </a> <a href="models.html" class="md-footer__link md-footer__link--next" aria-label="Next: nautobot.apps.models" rel="next"> <div class="md-footer__title"> Next <div class="md-ellipsis"> nautobot.apps.models </div> </div> <div class="md-footer__button md-icon"> <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24"><path d="M4 11v2h12l-5.5 5.5 1.42 1.42L19.84 12l-7.92-7.92L10.5 5.5 16 11H4Z"/></svg> </div> </a> </nav> <div class="md-footer-meta md-typeset"> <div class="md-footer-meta__inner md-grid"> <div class="md-copyright"> <div class="md-copyright__highlight"> Copyright © The Authors </div> <a href="https://github.com/nautobot/nautobot/blob/main/LICENSE.txt">Apache-2.0 LICENSE</a> </div> <div class="md-copyright"> <a href="https://squidfunk.github.io/mkdocs-material/" target="_blank" rel="noopener">Made with Material for MkDocs</a> | <a href="https://www.networktocode.com/community/" target="_blank" rel="noopener">Join Nautobot Slack</a> | <a href="https://networktocode.com" target="_blank" rel="noopener">Sponsored by <img src=../../../assets/networktocode_bw.png class="copyright-logo"></a> </div>  <div id="readthedocs-embed-flyout"></div> <div class="md-social"> <a href="https://blog.networktocode.com/blog/tags/nautobot" target="_blank" rel="noopener" title="Network to Code Blog" class="md-social__link"> <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 448 512"><path d="M0 64c0-17.7 14.3-32 32-32 229.8 0 416 186.2 416 416 0 17.7-14.3 32-32 32s-32-14.3-32-32C384 253.6 226.4 96 32 96 14.3 96 0 81.7 0 64zm0 352a64 64 0 1 1 128 0 64 64 0 1 1-128 0zm32-256c159.1 0 288 128.9 288 288 0 17.7-14.3 32-32 32s-32-14.3-32-32c0-123.7-100.3-224-224-224-17.7 0-32-14.3-32-32s14.3-32 32-32z"/></svg> </a> <a href="https://www.youtube.com/playlist?list=PLjA0bhxgryJ2Ts4GJMDA-tPzVWEncv4pb" target="_blank" rel="noopener" title="Nautobot Videos" class="md-social__link"> <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 576 512"><path d="M549.655 124.083c-6.281-23.65-24.787-42.276-48.284-48.597C458.781 64 288 64 288 64S117.22 64 74.629 75.486c-23.497 6.322-42.003 24.947-48.284 48.597-11.412 42.867-11.412 132.305-11.412 132.305s0 89.438 11.412 132.305c6.281 23.65 24.787 41.5 48.284 47.821C117.22 448 288 448 288 448s170.78 0 213.371-11.486c23.497-6.321 42.003-24.171 48.284-47.821 11.412-42.867 11.412-132.305 11.412-132.305s0-89.438-11.412-132.305zm-317.51 213.508V175.185l142.739 81.205-142.739 81.201z"/></svg> </a> <a href="https://www.networktocode.com/community/" target="_blank" rel="noopener" title="Network to Code Community" class="md-social__link"> <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 448 512"><path d="M94.12 315.1c0 25.9-21.16 47.06-47.06 47.06S0 341 0 315.1c0-25.9 21.16-47.06 47.06-47.06h47.06v47.06zm23.72 0c0-25.9 21.16-47.06 47.06-47.06s47.06 21.16 47.06 47.06v117.84c0 25.9-21.16 47.06-47.06 47.06s-47.06-21.16-47.06-47.06V315.1zm47.06-188.98c-25.9 0-47.06-21.16-47.06-47.06S139 32 164.9 32s47.06 21.16 47.06 47.06v47.06H164.9zm0 23.72c25.9 0 47.06 21.16 47.06 47.06s-21.16 47.06-47.06 47.06H47.06C21.16 243.96 0 222.8 0 196.9s21.16-47.06 47.06-47.06H164.9zm188.98 47.06c0-25.9 21.16-47.06 47.06-47.06 25.9 0 47.06 21.16 47.06 47.06s-21.16 47.06-47.06 47.06h-47.06V196.9zm-23.72 0c0 25.9-21.16 47.06-47.06 47.06-25.9 0-47.06-21.16-47.06-47.06V79.06c0-25.9 21.16-47.06 47.06-47.06 25.9 0 47.06 21.16 47.06 47.06V196.9zM283.1 385.88c25.9 0 47.06 21.16 47.06 47.06 0 25.9-21.16 47.06-47.06 47.06-25.9 0-47.06-21.16-47.06-47.06v-47.06h47.06zm0-23.72c-25.9 0-47.06-21.16-47.06-47.06 0-25.9 21.16-47.06 47.06-47.06h117.84c25.9 0 47.06 21.16 47.06 47.06 0 25.9-21.16 47.06-47.06 47.06H283.1z"/></svg> </a> <a href="https://github.com/nautobot/nautobot" target="_blank" rel="noopener" title="GitHub Repo" class="md-social__link"> <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 496 512"><path d="M165.9 397.4c0 2-2.3 3.6-5.2 3.6-3.3.3-5.6-1.3-5.6-3.6 0-2 2.3-3.6 5.2-3.6 3-.3 5.6 1.3 5.6 3.6zm-31.1-4.5c-.7 2 1.3 4.3 4.3 4.9 2.6 1 5.6 0 6.2-2s-1.3-4.3-4.3-5.2c-2.6-.7-5.5.3-6.2 2.3zm44.2-1.7c-2.9.7-4.9 2.6-4.6 4.9.3 2 2.9 3.3 5.9 2.6 2.9-.7 4.9-2.6 4.6-4.6-.3-1.9-3-3.2-5.9-2.9zM244.8 8C106.1 8 0 113.3 0 252c0 110.9 69.8 205.8 169.5 239.2 12.8 2.3 17.3-5.6 17.3-12.1 0-6.2-.3-40.4-.3-61.4 0 0-70 15-84.7-29.8 0 0-11.4-29.1-27.8-36.6 0 0-22.9-15.7 1.6-15.4 0 0 24.9 2 38.6 25.8 21.9 38.6 58.6 27.5 72.9 20.9 2.3-16 8.8-27.1 16-33.7-55.9-6.2-112.3-14.3-112.3-110.5 0-27.5 7.6-41.3 23.6-58.9-2.6-6.5-11.1-33.3 2.6-67.9 20.9-6.5 69 27 69 27 20-5.6 41.5-8.5 62.8-8.5s42.8 2.9 62.8 8.5c0 0 48.1-33.6 69-27 13.7 34.7 5.2 61.4 2.6 67.9 16 17.7 25.8 31.5 25.8 58.9 0 96.5-58.9 104.2-114.8 110.5 9.2 7.9 17 22.9 17 46.4 0 33.7-.3 75.4-.3 83.6 0 6.5 4.6 14.4 17.3 12.1C428.2 457.8 496 362.9 496 252 496 113.3 383.5 8 244.8 8zM97.2 352.9c-1.3 1-1 3.3.7 5.2 1.6 1.6 3.9 2.3 5.2 1 1.3-1 1-3.3-.7-5.2-1.6-1.6-3.9-2.3-5.2-1zm-10.8-8.1c-.7 1.3.3 2.9 2.3 3.9 1.6 1 3.6.7 4.3-.7.7-1.3-.3-2.9-2.3-3.9-2-.6-3.6-.3-4.3.7zm32.4 35.6c-1.6 1.3-1 4.3 1.3 6.2 2.3 2.3 5.2 2.6 6.5 1 1.3-1.3.7-4.3-1.3-6.2-2.2-2.3-5.2-2.6-6.5-1zm-11.4-14.7c-1.6 1-1.6 3.6 0 5.9 1.6 2.3 4.3 3.3 5.6 2.3 1.6-1.3 1.6-3.9 0-6.2-1.4-2.3-4-3.3-5.6-2z"/></svg> </a> <a href="https://twitter.com/networktocode" target="_blank" rel="noopener" title="Network to Code Twitter" class="md-social__link"> <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 512 512"><path d="M459.37 151.716c.325 4.548.325 9.097.325 13.645 0 138.72-105.583 298.558-298.558 298.558-59.452 0-114.68-17.219-161.137-47.106 8.447.974 16.568 1.299 25.34 1.299 49.055 0 94.213-16.568 130.274-44.832-46.132-.975-84.792-31.188-98.112-72.772 6.498.974 12.995 1.624 19.818 1.624 9.421 0 18.843-1.3 27.614-3.573-48.081-9.747-84.143-51.98-84.143-102.985v-1.299c13.969 7.797 30.214 12.67 47.431 13.319-28.264-18.843-46.781-51.005-46.781-87.391 0-19.492 5.197-37.36 14.294-52.954 51.655 63.675 129.3 105.258 216.365 109.807-1.624-7.797-2.599-15.918-2.599-24.04 0-57.828 46.782-104.934 104.934-104.934 30.213 0 57.502 12.67 76.67 33.137 23.715-4.548 46.456-13.32 66.599-25.34-7.798 24.366-24.366 44.833-46.132 57.827 21.117-2.273 41.584-8.122 60.426-16.243-14.292 20.791-32.161 39.308-52.628 54.253z"/></svg> </a> </div> </div> </div> </footer> </div> <div class="md-dialog" data-md-component="dialog"> <div class="md-dialog__inner md-typeset"></div> </div> <script id="__config" type="application/json">{"base": "../../..", "features": ["content.code.copy", "navigation.footer", "navigation.tabs", "navigation.tabs.sticky", "navigation.tracking", "search.highlight", "search.share", "search.suggest"], "search": "../../../assets/javascripts/workers/search.74e28a9f.min.js", "translations": {"clipboard.copied": "Copied to clipboard", "clipboard.copy": "Copy to clipboard", "search.result.more.one": "1 more on this page", "search.result.more.other": "# more on this page", "search.result.none": "No matching documents", "search.result.one": "1 matching document", "search.result.other": "# matching documents", "search.result.placeholder": "Type to start searching", "search.result.term.missing": "Missing", "select.version": "Select version"}}</script> <script src="../../../assets/javascripts/bundle.220ee61c.min.js"></script> <script src="https://code.jquery.com/jquery-3.6.0.min.js"></script> </body> </html>

nautobot.apps.jobs ¶

nautobot.apps.jobs.BaseJob ¶

Meta ¶

after_return(status, retval, task_id, args, kwargs, einfo) ¶

apply(args=None, kwargs=None, link=None, link_error=None, task_id=None, retries=None, throw=None, logfile=None, loglevel=None, headers=None, **options) ¶

as_form(data=None, files=None, initial=None, approval_view=False) classmethod ¶

as_form_class() classmethod ¶

before_start(task_id, args, kwargs) ¶

class_path() ¶

class_path_dotted() ¶

class_path_js_escaped() ¶

clear_cache() ¶

delete_files(*files_to_delete) ¶

deserialize_data(data) classmethod ¶

load_file(pk) staticmethod ¶

load_json(filename) ¶

load_yaml(filename) ¶

on_bound(app) classmethod ¶

on_failure(exc, task_id, args, kwargs, einfo) ¶

on_retry(exc, task_id, args, kwargs, einfo) ¶

on_success(retval, task_id, args, kwargs) ¶

prepare_job_kwargs(job_kwargs) classmethod ¶

properties_dict() ¶

run(*args, **kwargs) ¶

save_file(uploaded_file) staticmethod ¶

serialize_data(data) staticmethod ¶

shadow_name(args, kwargs, options) ¶

nautobot.apps.jobs.BooleanVar ¶

nautobot.apps.jobs.ChoiceVar ¶

nautobot.apps.jobs.DatabaseFileField ¶

nautobot.apps.jobs.DryRunVar ¶

nautobot.apps.jobs.FileVar ¶

nautobot.apps.jobs.GitRepositoryDryRun ¶

nautobot.apps.jobs.GitRepositorySync ¶

nautobot.apps.jobs.IPAddressVar ¶

nautobot.apps.jobs.IPAddressWithMaskVar ¶

nautobot.apps.jobs.IPNetworkVar ¶

nautobot.apps.jobs.IntegerVar ¶

nautobot.apps.jobs.Job ¶

nautobot.apps.jobs.JobButtonReceiver ¶

receive_job_button(obj) ¶

run(object_pk, object_model_name) ¶

nautobot.apps.jobs.JobHookReceiver ¶

receive_job_hook(change, action, changed_object) ¶

run(object_change) ¶

nautobot.apps.jobs.MultiChoiceVar ¶

nautobot.apps.jobs.MultiObjectVar ¶

nautobot.apps.jobs.NautobotKombuJSONEncoder ¶

nautobot.apps.jobs.ObjectVar ¶

nautobot.apps.jobs.RunJobTaskFailed ¶

nautobot.apps.jobs.ScriptVariable ¶

as_field() ¶

nautobot.apps.jobs.StringVar ¶

nautobot.apps.jobs.TextVar ¶

`nautobot.apps.jobs` ¶

`nautobot.apps.jobs.BaseJob` ¶

`Meta` ¶

`after_return(status, retval, task_id, args, kwargs, einfo)` ¶

`apply(args=None, kwargs=None, link=None, link_error=None, task_id=None, retries=None, throw=None, logfile=None, loglevel=None, headers=None, **options)` ¶

`as_form(data=None, files=None, initial=None, approval_view=False)` `classmethod` ¶

`as_form_class()` `classmethod` ¶

`before_start(task_id, args, kwargs)` ¶

`class_path()` ¶

`class_path_dotted()` ¶

`class_path_js_escaped()` ¶

`clear_cache()` ¶

`delete_files(*files_to_delete)` ¶

`deserialize_data(data)` `classmethod` ¶

`load_file(pk)` `staticmethod` ¶

`load_json(filename)` ¶

`load_yaml(filename)` ¶

`on_bound(app)` `classmethod` ¶

`on_failure(exc, task_id, args, kwargs, einfo)` ¶

`on_retry(exc, task_id, args, kwargs, einfo)` ¶

`on_success(retval, task_id, args, kwargs)` ¶

`prepare_job_kwargs(job_kwargs)` `classmethod` ¶

`properties_dict()` ¶

`run(*args, **kwargs)` ¶

`save_file(uploaded_file)` `staticmethod` ¶

`serialize_data(data)` `staticmethod` ¶

`shadow_name(args, kwargs, options)` ¶

`nautobot.apps.jobs.BooleanVar` ¶

`nautobot.apps.jobs.ChoiceVar` ¶

`nautobot.apps.jobs.DatabaseFileField` ¶

`nautobot.apps.jobs.DryRunVar` ¶

`nautobot.apps.jobs.FileVar` ¶

`nautobot.apps.jobs.GitRepositoryDryRun` ¶

`nautobot.apps.jobs.GitRepositorySync` ¶

`nautobot.apps.jobs.IPAddressVar` ¶

`nautobot.apps.jobs.IPAddressWithMaskVar` ¶

`nautobot.apps.jobs.IPNetworkVar` ¶

`nautobot.apps.jobs.IntegerVar` ¶

`nautobot.apps.jobs.Job` ¶

`nautobot.apps.jobs.JobButtonReceiver` ¶

`receive_job_button(obj)` ¶

`run(object_pk, object_model_name)` ¶

`nautobot.apps.jobs.JobHookReceiver` ¶

`receive_job_hook(change, action, changed_object)` ¶

`run(object_change)` ¶

`nautobot.apps.jobs.MultiChoiceVar` ¶

`nautobot.apps.jobs.MultiObjectVar` ¶

`nautobot.apps.jobs.NautobotKombuJSONEncoder` ¶

`nautobot.apps.jobs.ObjectVar` ¶

`nautobot.apps.jobs.RunJobTaskFailed` ¶

`nautobot.apps.jobs.ScriptVariable` ¶

`as_field()` ¶

`nautobot.apps.jobs.StringVar` ¶

`nautobot.apps.jobs.TextVar` ¶