Coverage for netbox/event_rules.py: 52%
69 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import re
3from django.core.exceptions import ImproperlyConfigured, ValidationError
4from django.utils.text import capfirst
5from django.utils.translation import gettext_lazy as _
7from netbox.registry import registry
8from utilities.choices import Choice
9from utilities.string import enum_key
11__all__ = (
12 'EventRuleAction',
13 'get_event_rule_action',
14 'get_event_rule_action_choices',
15 'get_event_rule_action_slugs',
16 'register_event_rule_action',
17)
19# A slug must sanitize (via enum_key()) into a valid GraphQL enum member name, which rules out a
20# leading digit or underscore. Hyphens are rejected outright rather than sanitized away, so that a
21# slug always reads as it was written.
22SLUG_RE = re.compile(r'^[a-z][a-z0-9_]*(\.[a-z0-9_]+)*$')
25# This module must not import any concrete Django models: it's imported by netbox.plugins (itself
26# imported by netbox.settings, before the app registry is populated), so only registry-level
27# bookkeeping belongs here. Subclasses that reference real models (e.g. NetBox's own
28# WebhookAction/ScriptAction/NotificationAction) live in netbox.extras.event_rules instead, and are
29# registered from ExtrasConfig.ready() once the app registry is available.
30class EventRuleAction:
31 """
32 Base class for a registered Event Rule action. Subclass this to add a new action type that an
33 EventRule can dispatch to, whether defined in NetBox core or in a plugin.
35 Registration instantiates the class once; that single instance serves every event rule, request,
36 and background worker thread for the lifetime of the process. Implementations must therefore be
37 stateless: enqueue() and validate() must not stash per-event data on self, as concurrent
38 dispatches would race over it. Everything an action needs is passed in as arguments.
40 Attributes:
41 slug: A unique identifier for this action (e.g. "webhook", or "myplugin.run_check" for a
42 plugin-provided action). Must begin with a lowercase letter, and may contain only
43 letters, digits, underscores, and dot-separated segments thereafter -- no hyphens (use
44 an underscore instead, e.g. "my_plugin.open_ticket"). A dotted namespace prefix is
45 strongly recommended for plugin-provided actions to avoid collisions with other plugins
46 or future core actions.
47 label: The human-friendly name shown in the UI/API.
48 description: An optional, longer description shown alongside the label (e.g. as a tooltip
49 in the action_type dropdown).
50 object_model: The model class (if any) which EventRule.action_object must be an instance of.
51 May be left as None if this action never operates against a target object, in which
52 case supplying an action_object is treated as a validation error.
53 object_required: Whether an action_object must be supplied for this action to be usable.
54 Defaults to False; set True (alongside object_model) when a target object is mandatory.
55 An action may declare object_model but still treat the object as optional.
56 object_label: The label for the object selection field on the event rule form. Defaults to
57 object_model's verbose name.
58 """
59 slug = None
60 label = None
61 description = None
62 object_model = None
63 object_required = False
64 object_label = None
66 # Set per-instance by register_event_rule_action(); a subclass override is ignored. Determines
67 # whether a dispatch-time exception from this action is isolated or propagates (see
68 # process_event_rules() in extras.events).
69 is_plugin_provided = True
71 def __repr__(self):
72 return f"<{self.__class__.__name__}: {self.slug}>"
74 def get_object_queryset(self):
75 """
76 Return the queryset of objects eligible for selection as this action's action_object, or
77 None if object_model is not set.
78 """
79 if self.object_model is None:
80 return None
81 return self.object_model.objects.all()
83 def get_object_label(self):
84 """
85 Return the label for this action's object selection field, or None if object_model is not
86 set. Defaults to object_model's verbose name, unless object_label has been set explicitly.
87 """
88 if self.object_label:
89 return self.object_label
90 if self.object_model is None:
91 return None
92 return capfirst(self.object_model._meta.verbose_name)
94 def resolve_import_object(self, value):
95 """
96 Optional hook: resolve a CSV/bulk-import "action object" string to a model instance. Raise
97 django.core.exceptions.ObjectDoesNotExist (or a subclass) if the value doesn't resolve.
98 Return None (the default) if this action doesn't support bulk import.
99 """
100 return
102 def _validate(self, *, action_object, action_data):
103 """
104 Entry point called from EventRule.clean(). Enforces the base object_required/object_model
105 checks, then delegates to validate() for any action-specific validation.
106 """
107 if self.object_required and action_object is None:
108 raise ValidationError({
109 'action_object_id': _("This action requires a target object to be selected."),
110 })
111 if action_object is not None:
112 if self.object_model is None:
113 raise ValidationError({
114 'action_object_id': _("This action does not operate against a target object."),
115 })
116 if not isinstance(action_object, self.object_model):
117 raise ValidationError({
118 'action_object_id': _("Selected object is not a valid {model}.").format(
119 model=self.object_model._meta.verbose_name
120 ),
121 })
122 self.validate(action_object=action_object, action_data=action_data)
124 def validate(self, *, action_object, action_data):
125 """
126 Optional hook: add custom validation, raising ValidationError on failure. No-op by
127 default; no need to call super() -- _validate() above runs the base checks regardless.
128 """
129 pass
131 def enqueue(self, *, event_rule, event_context, action_object, action_data):
132 """
133 Perform (or schedule) this action in response to a queued event. Implementations should
134 not raise for conditions that are the fault of this EventRule's own configuration alone;
135 log and return instead, so that other EventRules processed in the same batch are
136 unaffected.
137 """
138 raise NotImplementedError(f"{self.__class__.__name__} must implement enqueue()")
141def register_event_rule_action(cls, *, is_plugin_provided=True):
142 """
143 Register an EventRuleAction subclass. Can be used as a decorator, or called directly (e.g. when
144 iterating a plugin's declared event_rule_actions):
146 @register_event_rule_action
147 class MyAction(EventRuleAction):
148 slug = 'myplugin.my_action'
149 ...
151 Raises ImproperlyConfigured if slug/label are missing, the slug is malformed, already
152 registered, collides via enum_key() with another registered slug once both feed the GraphQL
153 EventRuleActionEnum (see extras.graphql.enums), or object_required is set without an
154 object_model to validate the object against. Checking slug/label here rather than at class
155 definition means an intermediate base class shared by several concrete actions in a plugin can
156 leave them unset.
158 is_plugin_provided determines whether a dispatch-time exception from this action is isolated
159 or propagates (see process_event_rules() in extras.events); defaults to True. NetBox's own
160 core registrations (extras.apps.ExtrasConfig) pass False explicitly.
161 """
162 instance = cls()
163 if not instance.slug: 163 ↛ 164line 163 didn't jump to line 164 because the condition on line 163 was never true
164 raise ImproperlyConfigured(f"{cls.__name__} must define a non-empty 'slug' attribute.")
165 if not instance.label: 165 ↛ 166line 165 didn't jump to line 166 because the condition on line 165 was never true
166 raise ImproperlyConfigured(f"{cls.__name__} must define a 'label' attribute.")
167 if instance.object_required and instance.object_model is None: 167 ↛ 169line 167 didn't jump to line 169 because the condition on line 167 was never true
168 # Unsatisfiable: an object is mandatory, yet any object supplied is rejected by _validate().
169 raise ImproperlyConfigured(
170 f"{cls.__name__} sets object_required but no object_model; a target object cannot be "
171 f"required for an action which does not operate against one."
172 )
173 if not SLUG_RE.fullmatch(instance.slug): 173 ↛ 174line 173 didn't jump to line 174 because the condition on line 173 was never true
174 raise ImproperlyConfigured(
175 f"Invalid event rule action slug {instance.slug!r}: must be lowercase, start with a "
176 f"letter, and use only letters, digits, underscores, and dot-separated segments."
177 )
178 if instance.slug in registry['event_rule_actions']: 178 ↛ 179line 178 didn't jump to line 179 because the condition on line 178 was never true
179 raise ImproperlyConfigured(f"An event rule action named {instance.slug} has already been registered!")
180 new_key = enum_key(instance.slug)
181 for existing in registry['event_rule_actions'].values():
182 if enum_key(existing.slug) == new_key: 182 ↛ 183line 182 didn't jump to line 183 because the condition on line 182 was never true
183 raise ImproperlyConfigured(
184 f"Event rule action slug {instance.slug!r} collides with the already-registered "
185 f"{existing.slug!r} once both are sanitized into a GraphQL enum member name."
186 )
187 instance.is_plugin_provided = is_plugin_provided
188 registry['event_rule_actions'][instance.slug] = instance
189 return cls
192def get_event_rule_action(slug):
193 return registry['event_rule_actions'].get(slug)
196def get_event_rule_action_choices():
197 return [
198 Choice(action.slug, action.label, description=action.description)
199 for action in registry['event_rule_actions'].values()
200 ]
203def get_event_rule_action_slugs():
204 return list(registry['event_rule_actions'].keys())