Initiating Nautobot jobs, next-2.0 rebase

changes/270.added

@@ -0,0 +1,4 @@
+Add init_jobs Nautobot subcommand.
+Add kwargs input to init_jobs as JSON-string to init_jobs.
+Add get_jobs Nautobot subcommand, which returns all Nautobot jobs viewable to user.
+Add filter_jobs Nautobot subcommand, which returns filtered set of Nautobot jobs viewable to user.

nautobot_chatops/workers/nautobot.py

@@ -1,5 +1,9 @@
 """Worker functions for interacting with Nautobot."""
 
+import json
+
+
+from django.contrib.auth import get_user_model
 from django.core.exceptions import ValidationError
 from django.db.models import Count
 from django.contrib.contenttypes.models import ContentType
@@ -10,7 +14,7 @@
 from nautobot.dcim.models import Device, DeviceType, Location, LocationType, Manufacturer, Rack, Cable
 from nautobot.ipam.models import VLAN, Prefix, VLANGroup
 from nautobot.tenancy.models import Tenant
-from nautobot.extras.models import Role, Status
+from nautobot.extras.models import Job, JobResult, Role, Status
 
 from nautobot_chatops.choices import CommandStatusChoices
 from nautobot_chatops.workers import subcommand_of, handle_subcommands
@@ -1045,6 +1049,140 @@ def get_circuit_providers(dispatcher, *args):
     return CommandStatusChoices.STATUS_SUCCEEDED
 
 
+@subcommand_of("nautobot")
+def filter_jobs(dispatcher, job_filters: str = ""):  # We can use a Literal["enabled", "installed"] here instead
+    """Get a filtered list of jobs from Nautobot that the request user have view permissions for.
+
+    Args:
+        job_filters (str): Filter job results by literals in a comma-separated string.
+                           Available filters are: enabled, installed.
+    """
+    # Check for filters in user supplied input
+    job_filters_list = [item.strip() for item in job_filters.split(",")] if isinstance(job_filters, str) else ""
+    filters = ["enabled", "installed"]
+    if any([key in job_filters for key in filters]):
+        filter_args = {key: True for key in filters if key in job_filters_list}
+        jobs = Job.objects.restrict(dispatcher.user, "view").filter(**filter_args)  # enabled=True, installed=True
+    else:
+        jobs = Job.objects.restrict(dispatcher.user, "view").all()
+
+    header = ["Name", "ID", "Enabled"]
+    rows = [
+        (
+            str(job.name),
+            str(job.id),
+            str(job.enabled),
+        )
+        for job in jobs
+    ]
+
+    dispatcher.send_large_table(header, rows)
+
+    return CommandStatusChoices.STATUS_SUCCEEDED
+
+
+@subcommand_of("nautobot")
+def get_jobs(dispatcher, kwargs: str = ""):
+    """Get all jobs from Nautobot that the requesting user have view permissions for.
+
+    Args:
+        kwargs (str): JSON-string array of header items to be exported.
+    """
+    # Confirm kwargs is valid JSON
+    json_args = ["Name", "Id", "Enabled"]
+    try:
+        if kwargs:
+            json_args = json.loads(kwargs)
+    except json.JSONDecodeError:
+        dispatcher.send_error(f"Invalid JSON-string, cannot decode: {kwargs}")
+        return (CommandStatusChoices.STATUS_FAILED, f"Invalid JSON-string, cannot decode: {kwargs}")
+
+    jobs = Job.objects.restrict(dispatcher.user, "view").all()
+
+    # Check if all items in json_args are valid keys (assuming all keys of job object are valid)
+    valid_keys = [attr for attr in dir(Job) if not callable(getattr(Job, attr)) and not attr.startswith("_")]
+    for item in json_args:
+        if item not in valid_keys:
+            dispatcher.send_error(f"Invalid item provided: {item}")
+            return (CommandStatusChoices.STATUS_FAILED, f"Invalid item provided: {item}")
+
+    # TODO: Check json_args are all valid keys
+    header = [item.capitalize() for item in json_args]
+    rows = [(tuple(str(getattr(job, item, "")) for item in json_args)) for job in jobs]
+
+    dispatcher.send_large_table(header, rows)
+
+    return CommandStatusChoices.STATUS_SUCCEEDED
+
+
+@subcommand_of("nautobot")
+def init_job(dispatcher, job_name: str, kwargs: str = ""):
+    """Initiate a job in Nautobot by job name.
+
+    Args:
+        job_name (str): Name of Nautobot job to run.
+        kwargs (str): JSON-string dictionary for input keyword arguments for job run.
+        #profile (str): Whether to profile the job execution.
+    """
+    # Confirm kwargs is valid JSON
+    json_args = {}
+    try:
+        if kwargs:
+            json_args = json.loads(kwargs)
+    except json.JSONDecodeError:
+        dispatcher.send_error(f"Invalid JSON-string, cannot decode: {kwargs}")
+        return (CommandStatusChoices.STATUS_FAILED, f"Invalid JSON-string, cannot decode: {kwargs}")
+
+    profile = False
+    if json_args.get("profile") and json_args["profile"] is True:
+        profile = True
+
+    # Get instance of the user who will run the job
+    user = get_user_model()
+    try:
+        user_instance = user.objects.get(username=dispatcher.user)
+    except user.DoesNotExist:  # Unsure if we need to check this case?
+        dispatcher.send_error(f"User {dispatcher.user} not found")
+        return (CommandStatusChoices.STATUS_FAILED, f'User "{dispatcher.user}" was not found')
+
+    # Get the job model instance using job name
+    try:
+        job_model = Job.objects.restrict(dispatcher.user, "view").get(name=job_name)
+    except Job.DoesNotExist:
+        dispatcher.send_error(f"Job {job_name} not found")
+        return (CommandStatusChoices.STATUS_FAILED, f'Job "{job_name}" was not found')
+
+    if not job_model.enabled:
+        dispatcher.send_error(f"The requested job {job_name} is not enabled")
+        return (CommandStatusChoices.STATUS_FAILED, f'Job "{job_name}" is not enabled')
+
+    job_class_path = job_model.class_path
+
+    # TODO: Check if json_args keys are valid for this job model
+    job_result = JobResult.execute_job(
+        job_model=job_model,
+        user=user_instance,
+        profile=profile,
+        **json_args,
+    )
+
+    if job_result and job_result.status == "FAILURE":
+        dispatcher.send_error(f"The requested job {job_name} failed to initiate. Result: {job_result.result}")
+        return (CommandStatusChoices.STATUS_FAILED, f'Job "{job_name}" failed to initiate. Result: {job_result.result}')
+
+    # TODO: need base-domain, this yields: /extras/job-results/<job_id>/
+    job_url = job_result.get_absolute_url()
+    blocks = [
+        dispatcher.markdown_block(
+            f"The requested job {job_class_path} was initiated! [`click here`]({job_url}) to open the job."
+        ),
+    ]
+
+    dispatcher.send_blocks(blocks)
+
+    return CommandStatusChoices.STATUS_SUCCEEDED
+
+
 @subcommand_of("nautobot")
 def about(dispatcher, *args):
     """Provide link for more information on Nautobot Apps."""