Coverage for src/backend/InvenTree/common/notifications.py: 70%
94 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
1"""Base classes and functions for notifications."""
3from dataclasses import dataclass
4from datetime import timedelta
5from typing import Optional
7from django.contrib.auth import get_user_model
8from django.contrib.auth.models import Group
9from django.db.models import Model
10from django.utils.translation import gettext_lazy as _
12import structlog
14import common.models
15from InvenTree.exceptions import log_error
16from InvenTree.ready import isImportingData, isRebuildingData
17from plugin import PluginMixinEnum, registry
18from users.models import Owner
19from users.permissions import check_user_permission
21logger = structlog.get_logger('inventree')
24# region methods
25@dataclass()
26class NotificationBody:
27 """Information needed to create a notification.
29 Attributes:
30 name (str): Name (or subject) of the notification
31 slug (str): Slugified reference for notification
32 message (str): Notification message as text. Should not be longer than 120 chars.
33 template (str): Reference to the html template for the notification.
35 The strings support f-string style formatting with context variables parsed at runtime.
37 Context variables:
38 instance: Text representing the instance
39 verbose_name: Verbose name of the model
40 app_label: App label (slugified) of the model
41 model_name': Name (slugified) of the model
42 """
44 name: str
45 slug: str
46 message: str
47 template: Optional[str] = None
50class InvenTreeNotificationBodies:
51 """Default set of notifications for InvenTree.
53 Contains regularly used notification bodies.
54 """
56 NewOrder = NotificationBody(
57 name=_('New {verbose_name}'),
58 slug='{app_label}.new_{model_name}',
59 message=_('A new order has been created and assigned to you'),
60 template='email/new_order_assigned.html',
61 )
62 """Send when a new order (build, sale or purchase) was created."""
64 OrderCanceled = NotificationBody(
65 name=_('{verbose_name} canceled'),
66 slug='{app_label}.canceled_{model_name}',
67 message=_('A order that is assigned to you was canceled'),
68 template='email/canceled_order_assigned.html',
69 )
70 """Send when a order (sale, return or purchase) was canceled."""
72 ItemsReceived = NotificationBody(
73 name=_('Items Received'),
74 slug='purchase_order.items_received',
75 message=_('Items have been received against a purchase order'),
76 template='email/purchase_order_received.html',
77 )
79 ReturnOrderItemsReceived = NotificationBody(
80 name=_('Items Received'),
81 slug='return_order.items_received',
82 message=_('Items have been received against a return order'),
83 template='email/return_order_received.html',
84 )
87def trigger_notification(obj: Model, category: str = '', obj_ref: str = 'pk', **kwargs):
88 """Send out a notification.
90 Args:
91 obj: The object (model instance) that is triggering the notification
92 category: The category (label) for the notification
93 obj_ref: The reference to the object that should be used for the notification
94 kwargs: Additional arguments to pass to the notification method
95 """
96 # Check if data is importing currently
97 if isImportingData() or isRebuildingData(): # pragma: no cover 97 ↛ 98line 97 didn't jump to line 98 because the condition on line 97 was never true
98 return
100 targets = kwargs.get('targets')
101 target_fnc = kwargs.get('target_fnc')
102 target_args = kwargs.get('target_args', [])
103 target_kwargs = kwargs.get('target_kwargs', {})
104 target_exclude = kwargs.get('target_exclude')
105 context = kwargs.get('context', {})
106 delivery_methods = kwargs.get('delivery_methods')
107 check_recent = kwargs.get('check_recent', True)
109 # Resolve object reference
110 refs = [obj_ref, 'pk', 'id', 'uid']
112 obj_ref_value = None
114 # Find the first reference that is available
115 if obj: 115 ↛ 127line 115 didn't jump to line 127 because the condition on line 115 was always true
116 for ref in refs: 116 ↛ 121line 116 didn't jump to line 121 because the loop on line 116 didn't complete
117 if hasattr(obj, ref): 117 ↛ 116line 117 didn't jump to line 116 because the condition on line 117 was always true
118 obj_ref_value = getattr(obj, ref)
119 break
121 if not obj_ref_value: 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true
122 raise KeyError(
123 f"Could not resolve an object reference for '{obj!s}' with {','.join(set(refs))}"
124 )
126 # Check if we have notified recently...
127 delta = timedelta(days=1)
129 if check_recent and common.models.NotificationEntry.check_recent( 129 ↛ 132line 129 didn't jump to line 132 because the condition on line 129 was never true
130 category, obj_ref_value, delta
131 ):
132 logger.info(
133 "Notification '%s' has recently been sent for '%s' - SKIPPING",
134 category,
135 obj,
136 )
137 return
139 logger.info("Gathering users for notification '%s'", category)
141 if target_exclude is None:
142 target_exclude = set()
144 # Collect possible targets
145 if not targets and target_fnc: 145 ↛ 146line 145 didn't jump to line 146 because the condition on line 145 was never true
146 targets = target_fnc(*target_args, **target_kwargs)
148 # Convert list of targets to a list of users
149 # (targets may include 'owner' or 'group' classes)
150 target_users = set()
152 if targets: 152 ↛ 178line 152 didn't jump to line 178 because the condition on line 152 was always true
153 for target in targets:
154 if target is None:
155 continue
156 # User instance is provided
157 elif isinstance(target, get_user_model()): 157 ↛ 161line 157 didn't jump to line 161 because the condition on line 157 was always true
158 if target not in target_exclude: 158 ↛ 153line 158 didn't jump to line 153 because the condition on line 158 was always true
159 target_users.add(target)
160 # Group instance is provided
161 elif isinstance(target, Group):
162 for user in get_user_model().objects.filter(groups__name=target.name):
163 if user not in target_exclude:
164 target_users.add(user)
165 # Owner instance (either 'user' or 'group' is provided)
166 elif isinstance(target, Owner):
167 for owner in target.get_related_owners(include_group=False):
168 user = owner.owner
169 if user not in target_exclude:
170 target_users.add(user)
171 # Unhandled type
172 else:
173 logger.error(
174 'Unknown target passed to trigger_notification method: %s', target
175 )
177 # Filter out any users who are inactive, or do not have the required model permissions
178 valid_users = list(
179 filter(
180 lambda u: (
181 u and u.is_active and (not obj or check_user_permission(u, obj, 'view'))
182 ),
183 list(target_users),
184 )
185 )
187 # Track whether any notifications were sent
188 result = False
190 # Send out via all registered notification methods
191 for plugin in registry.with_mixin(PluginMixinEnum.NOTIFICATION):
192 # Skip if the plugin is *not* in the "delivery_methods" list?
193 match = not delivery_methods
195 for notification_class in delivery_methods or []:
196 if type(notification_class) is str: 196 ↛ 197line 196 didn't jump to line 197 because the condition on line 196 was never true
197 if plugin.slug == notification_class:
198 match = True
199 break
201 elif getattr(notification_class, 'SLUG', None) == plugin.slug:
202 match = True
203 break
205 if not match:
206 continue
208 try:
209 # Plugin may optionally filter target users
210 filtered_users = plugin.filter_targets(list(valid_users))
211 if plugin.send_notification(obj, category, filtered_users, context):
212 result = True
213 except Exception:
214 log_error('send_notification', plugin=plugin.slug)
216 # Log the notification entry
217 if result:
218 common.models.NotificationEntry.notify(category, obj_ref_value)