Notification Plugin - How to

[Bbillyben]

Hi There,

I would dev a plugin to [daily/weekly/userdefine] be notified of stock stale (and/or other stock management figures).

I would use the core notification process. and I'm currently into self feasibility assessment, so only ideas no script yet.

I'm trying to figure out how it's done for current event notification (eg stock under minimum level, stock expiricy), but I can't find how a new message is added, how the target users are selected.

I've seen the comon.notification.NotificationMethod / BulkNotificationMethod where 'based' targets are defined in init
and the core plugin plugin.builtin.integration.core_notification.CoreNotificationsPlugin + EmailNotification
but I can't find where it's being instanciated (so how is defined the self.targets...)

Could you point me where to start digging into the notification process through inventree ?

[matmair]

Hi @Bbillyben!

CoreNotificationMethod etc. define the delivery methods that can be used by InvenTree. Every time common.notifications.trigger_notification gets called all notification methods (which are classes but are named NotificationMethod) are called in order of registration and get passed a target (user) list. This list is filtered (some notification methods might only delivery to a subset of users) and then the send method is called to initiate delivery.

Similar features to what you are proposing are already present so this should be doable.
There are a few steps to this, I will link some sample code:

1. Create a function in the plugin that gets scheduled - use the ScheduleMixin for this
   

   InvenTree/InvenTree/plugin/samples/integration/scheduled_task.py

   Lines 24 to 62 in e730b5c

   	class ScheduledTaskPlugin(ScheduleMixin, SettingsMixin, InvenTreePlugin):
   	"""A sample plugin which provides support for scheduled tasks."""
   	NAME = "ScheduledTasksPlugin"
   	SLUG = "schedule"
   	TITLE = "Scheduled Tasks"
   	SCHEDULED_TASKS = {
   	'member': {
   	'func': 'member_func',
   	'schedule': 'I',
   	'minutes': 30,
   	},
   	'hello': {
   	'func': 'plugin.samples.integration.scheduled_task.print_hello',
   	'schedule': 'I',
   	'minutes': 45,
   	},
   	'world': {
   	'func': 'plugin.samples.integration.scheduled_task.print_world',
   	'schedule': 'H',
   	},
   	}
   	SETTINGS = {
   	'T_OR_F': {
   	'name': 'True or False',
   	'description': 'Print true or false when running the periodic task',
   	'validator': bool,
   	'default': False,
   	},
   	}
   	def member_func(self, *args, **kwargs):
   	"""A simple member function to demonstrate functionality."""
   	t_or_f = self.get_setting('T_OR_F')
   	print(f"Called member_func - value is {t_or_f}")
   	return t_or_f

2. The function should check for the conditions you use to determine if you should send a notification out (maybe make them concrete - one function per condition, one caller function that calls all the concrete ones)
   

   InvenTree/InvenTree/build/tasks.py

   Lines 37 to 77 in e730b5c

   	lines = []
   	if not build:
   	logger.error("Invalid build passed to 'build.tasks.check_build_stock'")
   	return
   	try:
   	part = build.part
   	except part_models.Part.DoesNotExist:
   	# Note: This error may be thrown during unit testing...
   	logger.error("Invalid build.part passed to 'build.tasks.check_build_stock'")
   	return
   	for bom_item in part.get_bom_items():
   	sub_part = bom_item.sub_part
   	# The 'in stock' quantity depends on whether the bom_item allows variants
   	in_stock = sub_part.get_stock_count(include_variants=bom_item.allow_variants)
   	allocated = sub_part.allocation_count()
   	available = max(0, in_stock - allocated)
   	required = Decimal(bom_item.quantity) * Decimal(build.quantity)
   	if available < required:
   	# There is not sufficient stock for this part
   	lines.append({
   	'link': InvenTree.helpers.construct_absolute_url(sub_part.get_absolute_url()),
   	'part': sub_part,
   	'in_stock': in_stock,
   	'allocated': allocated,
   	'available': available,
   	'required': required,
   	})
   	if len(lines) == 0:
   	# Nothing to do
   	return

3. Determine who should get the notification
   

   InvenTree/InvenTree/part/models.py

   Lines 260 to 278 in e730b5c

   	def get_subscribers(self, include_parents=True):
   	"""Return a list of users who subscribe to this PartCategory."""
   	cats = self.get_ancestors(include_self=True)
   	subscribers = set()
   	if include_parents:
   	queryset = PartCategoryStar.objects.filter(
   	category__in=cats,
   	)
   	else:
   	queryset = PartCategoryStar.objects.filter(
   	category=self,
   	)
   	for result in queryset:
   	subscribers.add(result.user)
   	return [s for s in subscribers]

4. Send the notification
   

   InvenTree/InvenTree/part/tasks.py

   Lines 19 to 43 in e730b5c

   	def notify_low_stock(part: part.models.Part):
   	"""Notify interested users that a part is 'low stock':
   	- Triggered when the available stock for a given part falls be low the configured threhsold
   	- A notification is delivered to any users who are 'subscribed' to this part
   	"""
   	name = _("Low stock notification")
   	message = _(f'The available stock for {part.name} has fallen below the configured minimum level')
   	context = {
   	'part': part,
   	'name': name,
   	'message': message,
   	'link': InvenTree.helpers.construct_absolute_url(part.get_absolute_url()),
   	'template': {
   	'html': 'email/low_stock_notification.html',
   	'subject': name,
   	},
   	}
   	common.notifications.trigger_notification(
   	part,
   	'part.notify_low_stock',
   	target_fnc=part.get_subscribers,
   	context=context,
   	)

[Bbillyben]

Thanks @matmair !

I can start digging the stream!

I can see the instanciation of NotificationMethod in def deliver_notification for instance.

[matmair]

@Bbillyben you can just ship a template with your plugin, should work out of the box