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

14132 review changes fix tests

Browse Source
This commit is contained in:
Arthur 2023-11-21 08:45:38 -08:00
parent 56a9210d6c
commit 7eb5371600
3 changed files with 5 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
netbox/extras/events.py
View File

@ -3,9 +3,14 @@ import sys
from django.conf import settings from django.conf import settings
from django.contrib.contenttypes.models import ContentType from django.contrib.contenttypes.models import ContentType
from django_rq import get_queue
from django.utils import timezone
from netbox.config import get_config
from netbox.constants import RQ_QUEUE_DEFAULT
from netbox.registry import registry from netbox.registry import registry
from utilities.api import get_serializer_for_model from utilities.api import get_serializer_for_model
from utilities.rqworker import get_rq_retry
from utilities.utils import serialize_object from utilities.utils import serialize_object
from .choices import * from .choices import *
from .models import EventRule from .models import EventRule

1
netbox/extras/models/models.py
View File

@ -170,7 +170,6 @@ class EventRule(CustomFieldsMixin, ExportTemplatesMixin, TagsMixin, ChangeLogged
if not self.conditions: if not self.conditions:
return True return True
logger.debug(f'Evaluating event rule conditions: {self.conditions}')
if ConditionSet(self.conditions).eval(data): if ConditionSet(self.conditions).eval(data):
return True return True

38
netbox/extras/utils.py
View File

@ -1,16 +1,12 @@
import logging import logging
from django.db.models import Q from django.db.models import Q
from django_rq import get_queue
from django.utils import timezone
from django.utils.deconstruct import deconstructible from django.utils.deconstruct import deconstructible
from taggit.managers import _TaggableManager from taggit.managers import _TaggableManager
from extras.conditions import ConditionSet from extras.conditions import ConditionSet
from netbox.constants import RQ_QUEUE_DEFAULT
from extras.choices import EventRuleActionChoices from extras.choices import EventRuleActionChoices
from netbox.config import get_config from netbox.config import get_config
from netbox.registry import registry from netbox.registry import registry
from utilities.rqworker import get_rq_retry
logger = logging.getLogger('netbox.extras.utils') logger = logging.getLogger('netbox.extras.utils')
@ -79,37 +75,3 @@ def is_report(obj):
return issubclass(obj, Report) and obj != Report return issubclass(obj, Report) and obj != Report
except TypeError: except TypeError:
return False return False
def process_event_rules(event_rules, model_name, event, data, username, snapshots=None, request_id=None):
rq_queue_name = get_config().QUEUE_MAPPINGS.get('webhook', RQ_QUEUE_DEFAULT)
rq_queue = get_queue(rq_queue_name)
for event_rule in event_rules:
if event_rule.action_type == EventRuleActionChoices.WEBHOOK:
processor = "extras.webhooks_worker.process_webhook"
elif event_rule.action_type == EventRuleActionChoices.SCRIPT:
processor = "extras.scripts_worker.process_script"
else:
raise ValueError(f"Unknown action type for an event rule: {event_rule.action_type}")
params = {
"event_rule": event_rule,
"model_name": model_name,
"event": event,
"data": data,
"snapshots": snapshots,
"timestamp": str(timezone.now()),
"username": username,
"retry": get_rq_retry()
}
if snapshots:
params["snapshots"] = snapshots
if request_id:
params["request_id"] = request_id
rq_queue.enqueue(
processor,
**params
)