Next 2.0

changes/270.added

@@ -0,0 +1,4 @@
+Add run_job Nautobot subcommand, which initiates a job with kwargs or a job requiring no manual form input.
+Add run_job_form Nautobot subcommand, which presents job's form widgets to the user.
+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
+import time
+
+
 from django.core.exceptions import ValidationError
 from django.db.models import Count
 from django.contrib.contenttypes.models import ContentType
@@ -10,7 +14,9 @@
 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.choices import JobResultStatusChoices
+from nautobot.extras.models import Job, JobResult, Role, Status
+from nautobot.extras.jobs import get_job
 
 from nautobot_chatops.choices import CommandStatusChoices
 from nautobot_chatops.workers import subcommand_of, handle_subcommands
@@ -1045,6 +1051,346 @@ 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. (Optional, default export is: name, id, enabled)
+    """
+    # 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}")
+
+    # Confirm `name` is always present in export
+    name_key = json_args.get("name") or json_args.get("Name")
+    if not name_key:
+        json_args.append("name")
+
+    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.lower() not in valid_keys:
+            dispatcher.send_error(f"Invalid item provided: {item.lower()}")
+            return (CommandStatusChoices.STATUS_FAILED, f"Invalid item provided: {item.lower()}")
+
+    # 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 run_job(dispatcher, *args, job_name: str = "", json_string_kwargs: str = ""):  # pylint: disable=too-many-locals
+    """Initiate a job in Nautobot by job name.
+
+    Args:
+        *args (tuple): Dispatcher form will pass job args as tuple.
+        job_name (str): Name of Nautobot job to run.
+        json_string_kwargs (str): JSON-string dictionary for input keyword arguments for job run.
+    """
+    # Prompt the user to pick a job if they did not specify one
+    if not job_name:
+        return prompt_for_job(dispatcher, "nautobot run-job")
+
+    if args:
+        json_string_kwargs = "{}"
+
+    # Confirm kwargs is valid JSON
+    json_args = {}
+    try:
+        if json_string_kwargs:
+            json_args = json.loads(json_string_kwargs)
+    except json.JSONDecodeError:
+        dispatcher.send_error(f"Invalid JSON-string, cannot decode: {json_string_kwargs}")
+        return (CommandStatusChoices.STATUS_FAILED, f"Invalid JSON-string, cannot decode: {json_string_kwargs}")
+
+    profile = False
+    if json_args.get("profile") and json_args["profile"] is True:
+        profile = True
+
+    # 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')
+
+    form_class = get_job(job_model.class_path).as_form()
+
+    # Parse base form fields from job class
+    form_fields = []
+    for field_name, _ in form_class.base_fields.items():  # pylint: disable=unused-variable
+        if field_name.startswith("_"):
+            continue
+        form_fields.append(f"{field_name}")
+
+    # Basic logic check with what we know, we should expect run-job-form vs run-job to parse the same base fields
+    if len(form_fields) != len(args):
+        dispatcher.send_error("The form class fields and the passed run-job args do not match.")
+        return (
+            CommandStatusChoices.STATUS_FAILED,
+            "The form class fields and the passed run-job args do not match.",
+        )
+
+    form_item_kwargs = {}
+    for index, _ in enumerate(form_fields):  # pylint: disable=unused-variable
+        # Check request input (string-type) is also valid JSON
+        if args[index][0] == "{":
+            try:
+                json_arg = json.loads(args[index])
+                if not json_arg.get("id"):
+                    dispatcher.send_error("Form field arg is JSON dictionary, and has no `id` key.")
+                    return (
+                        CommandStatusChoices.STATUS_FAILED,
+                        "Form field arg is JSON dictionary, and has no `id` key.",
+                    )
+                form_item_kwargs[form_fields[index]] = json_arg.get("id")
+                continue
+            except json.JSONDecodeError:
+                form_item_kwargs[form_fields[index]] = args[index]
+                continue
+        form_item_kwargs[form_fields[index]] = args[index]
+
+    job_result = JobResult.enqueue_job(
+        job_model=job_model,
+        user=dispatcher.user,
+        profile=profile,
+        **form_item_kwargs,
+    )
+
+    # Wait on the job to finish
+    max_wait_iterations = 60
+    while job_result.status not in JobResultStatusChoices.READY_STATES:
+        max_wait_iterations -= 1
+        if not max_wait_iterations:
+            dispatcher.send_error(f"The requested job {job_name} failed to reach ready state.")
+            return (CommandStatusChoices.STATUS_FAILED, f'Job "{job_name}" failed to reach ready state.')
+        time.sleep(1)
+        job_result.refresh_from_db()
+
+    if job_result and job_result.status == "FAILURE":
+        dispatcher.send_error(f"The requested job {job_name} was initiated but failed. Result: {job_result.result}")
+        return (
+            CommandStatusChoices.STATUS_FAILED,
+            f'Job "{job_name}" was initiated but failed. Result: {job_result.result}',
+        )  # pylint: disable=line-too-long
+
+    job_url = (
+        f"{dispatcher.context['request_scheme']}://{dispatcher.context['request_host']}{job_result.get_absolute_url()}"
+    )
+    blocks = [
+        dispatcher.markdown_block(
+            f"The requested job {job_model.class_path} was initiated! [`click here`]({job_url}) to open the job."
+        ),
+    ]
+
+    dispatcher.send_blocks(blocks)
+
+    return CommandStatusChoices.STATUS_SUCCEEDED
+
+
+@subcommand_of("nautobot")
+def run_job_form(dispatcher, job_name: str = ""):
+    """Send job form as a multi-input dialog. On form submit it initiates the job with the form arguments.
+
+    Args:
+        job_name (str): Name of Nautobot job to run.
+    """
+    # Prompt the user to pick a job if they did not specify one
+    if not job_name:
+        return prompt_for_job(dispatcher, "nautobot run-job-form")
+
+    # Get jobs available to user
+    try:
+        job = Job.objects.restrict(dispatcher.user, "view").get(name=job_name)
+    except Job.DoesNotExist:
+        blocks = [
+            dispatcher.markdown_block(
+                f"Job {job_name} does not exist or {dispatcher.user} does not have permissions to run job."  # pylint: disable=line-too-long
+            ),
+        ]
+        dispatcher.send_blocks(blocks)
+        return CommandStatusChoices.STATUS_SUCCEEDED
+
+    except Job.MultipleObjectsReturned:
+        blocks = [
+            dispatcher.markdown_block(f"Multiple jobs found by name {job_name}."),
+        ]
+        dispatcher.send_blocks(blocks)
+        return CommandStatusChoices.STATUS_SUCCEEDED
+
+    if not job.enabled:
+        blocks = [
+            dispatcher.markdown_block(f"Job {job_name} is not enabled. The job must be enabled to be ran."),
+        ]
+        dispatcher.send_blocks(blocks)
+        return CommandStatusChoices.STATUS_SUCCEEDED
+
+    form_class = get_job(job.class_path).as_form()
+
+    # Parse base form fields from job class
+    form_items = {}
+    for field_name, field in form_class.base_fields.items():
+        if field_name.startswith("_"):
+            continue
+        form_items[field_name] = field
+
+    form_item_dialogs = []
+    for field_name, field in form_items.items():
+        try:
+            field_type = field.widget.input_type
+        except AttributeError:
+            # Some widgets (eg: textarea) do have the `input_type` attribute
+            field_type = field.widget.template_name.split("/")[-1].split(".")[0]
+
+        if field_type == "select":
+            if not hasattr(field, "choices"):
+                blocks = [
+                    dispatcher.markdown_block(f"Job {job_name} field {field} has no attribute `choices`."),
+                ]
+                dispatcher.send_blocks(blocks)
+                return CommandStatusChoices.STATUS_SUCCEEDED
+
+            query_result_items = []
+            for choice, value in field.choices:
+                query_result_items.append(
+                    (value, f'{{"field_name": "{field_name}", "value": "{value}", "id": "{str(choice)}"}}')
+                )
+
+            if len(query_result_items) == 0 and field.required:
+                blocks = [
+                    dispatcher.markdown_block(
+                        f"Job {job_name} for {field_name} is required, however no choices populated for dialog choices."
+                    ),
+                ]
+                dispatcher.send_blocks(blocks)
+                return CommandStatusChoices.STATUS_SUCCEEDED
+
+            form_item_dialogs.append(
+                {
+                    "type": field_type,
+                    "label": f"{field_name}: {field.help_text}",
+                    "choices": query_result_items,
+                    "default": query_result_items[0] if query_result_items else ("", ""),
+                    "confirm": False,
+                }
+            )
+
+        elif field_type == "text":
+            default_value = field.initial
+            form_item_dialogs.append(
+                {
+                    "type": field_type,
+                    "label": f"{field_name}: {field.help_text}",
+                    "default": default_value,
+                    "confirm": False,
+                }
+            )
+
+        elif field_type == "number":
+            # TODO: Can we enforce numeric-character mask for widget input without JavaScript?
+            default_value = field.initial
+            form_item_dialogs.append(
+                {
+                    "type": "text",
+                    "label": f"{field_name}: {field.help_text} *integer values only*",
+                    "default": default_value,
+                    "confirm": False,
+                }
+            )
+
+        elif field_type == "checkbox":
+            # TODO: Is there a checkbox widget?
+            default_value = ("False", "false")
+            if field.initial:
+                default_value = ("True", "true")
+            form_item_dialogs.append(
+                {
+                    "type": "select",
+                    "label": f"{field_name}: {field.help_text}",
+                    "choices": [("True", "true"), ("False", "false")],
+                    "default": default_value,
+                    "confirm": False,
+                }
+            )
+
+        elif field_type == "textarea":
+            # TODO: Is there a multi-line text input widget
+            default_value = field.initial
+            form_item_dialogs.append(
+                {
+                    "type": "text",
+                    "label": f"{field_name}: {field.help_text}",
+                    "default": default_value,
+                    "confirm": False,
+                }
+            )
+
+    # TODO: BUG: Single inputs will present but not submit properly with multi_input_dialog
+    dispatcher.multi_input_dialog(
+        command="nautobot",
+        sub_command=f"run-job {job_name} {{}}",
+        dialog_title=f"job {job_name} form input",
+        dialog_list=form_item_dialogs,
+    )
+
+    return CommandStatusChoices.STATUS_SUCCEEDED
+
+
+def prompt_for_job(dispatcher, command):
+    """Prompt the user to select a Nautobot Job."""
+    jobs = Job.objects.restrict(dispatcher.user, "view").all()
+    dispatcher.prompt_from_menu(command, "Select a Nautobot Job", [(job.name, job.name) for job in jobs])
+    return False
+
+
 @subcommand_of("nautobot")
 def about(dispatcher, *args):
     """Provide link for more information on Nautobot Apps."""