NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-08-23 16:06:43 -06:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

14132 script run fixup

Browse Source
This commit is contained in:
Arthur 2023-11-13 15:53:00 -08:00
parent f5615e5c8a
commit 3dbdd47ada
3 changed files with 27 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
netbox/extras/events_worker.py
View File

@ -45,7 +45,7 @@ def module_member(name):
def process_event_rules(event_rule, model_name, event, data, timestamp, username, request_id, snapshots): def process_event_rules(event_rule, model_name, event, data, timestamp, username, request_id, snapshots):
if event_rule.action_type == EventRuleActionChoices.WEBHOOK: if event_rule.action_type == EventRuleActionChoices.WEBHOOK:
process_webhook(event_rule, model_name, event, data, timestamp, username, request_id, snapshots) process_webhook(event_rule, model_name, event, data, timestamp, username, request_id, snapshots)
elif event_rule.action_type == EventRuleActionChoicesEventRuleTypeChoices.SCRIPT: elif event_rule.action_type == EventRuleActionChoices.SCRIPT:
process_script(event_rule, model_name, event, data, timestamp, username, request_id, snapshots) process_script(event_rule, model_name, event, data, timestamp, username, request_id, snapshots)

2
netbox/extras/forms/model_forms.py
View File

@ -293,7 +293,7 @@ class EventRuleForm(NetBoxModelForm):
def get_webhook_choices(self): def get_webhook_choices(self):
initial = None initial = None
if self.fields['action_object_type'] and self.fields['action_object_id']: if self.fields['action_object_type'] and get_field_value(self, 'action_object_id'):
initial = Webhook.objects.get(pk=get_field_value(self, 'action_object_id')) initial = Webhook.objects.get(pk=get_field_value(self, 'action_object_id'))
self.fields['action_choice'] = DynamicModelChoiceField( self.fields['action_choice'] = DynamicModelChoiceField(
label=_('Webhook'), label=_('Webhook'),

31
netbox/extras/scripts_worker.py
View File

@ -1,20 +1,39 @@
import logging import logging
import requests import requests
from core.models import Job
from django.conf import settings from django.conf import settings
from django_rq import job from django_rq import job
from jinja2.exceptions import TemplateError from jinja2.exceptions import TemplateError
from utilities.rqworker import get_workers_for_queue
from .conditions import ConditionSet from extras.conditions import ConditionSet
from .constants import WEBHOOK_EVENT_TYPES from extras.constants import WEBHOOK_EVENT_TYPES
from .webhooks import generate_signature from extras.models import ScriptModule
from extras.webhooks import generate_signature
logger = logging.getLogger('netbox.webhooks_worker') logger = logging.getLogger('netbox.webhooks_worker')
def process_script(webhook, model_name, event, data, timestamp, username, request_id=None, snapshots=None): def process_script(event_rule, model_name, event, data, timestamp, username, request_id=None, snapshots=None):
""" """
Make a POST request to the defined Webhook Run the requested script
""" """
module_name = event_rule.action_object_identifier.split(":")[0]
script_name = event_rule.action_object_identifier.split(":")[1]
try:
module = ScriptModule.objects.get(file_path__regex=f"^{module_name}\\.")
except ScriptModule.DoesNotExist:
return
pass script = module.scripts[script_name]()
job = Job.enqueue(
run_script,
instance=module,
name=script.class_name,
user=None,
schedule_at=None,
interval=None,
data=event_rule.parameters,
)