InvenTree 零件通知系统详解：低库存预警、订阅机制与源码级实现原理

InvenTree 零件通知系统详解：低库存预警、订阅机制与源码级实现原理

【免费下载链接】InvenTreeOpen Source Inventory Management System项目地址: https://gitcode.com/GitHub_Trending/in/InvenTree

本文以 InvenTree 官方文档docs/docs/part/notification.md为主体，系统讲解 InvenTree 中面向"零件（Part）"维度的通知机制：用户如何通过邮件和界面接收通知、低库存通知与生产工单通知的触发条件、如何订阅零件或零件分类，并结合开源仓库中的源码（信号钩子、后台任务、去重模型与插件分发管线）还原整条通知链路，帮助读者不仅会用该功能，还能理解其底层调用关系与防骚扰设计。

通知的总体工作方式与前提条件

InvenTree 允许用户在特定事件发生时接收通知。文档明确指出两个前提：

• 必须正确配置邮件服务：外部（邮件）通知依赖 邮件配置，并且需要在Notifications设置中启用通知功能；
• 每个用户必须绑定有效邮箱：否则无法收到邮件通知。

除邮件渠道外，通知还会直接展示在用户界面中：

1. 页头提示：有新通知时，页面头部会以角标形式提示；
2. 下拉飞览（flyout）：点击头部通知图标可展开查看最新通知；
3. 收件箱（Inbox）：列出当前所有通知；
4. 历史（History）：列出所有历史通知，支持逐条删除或一次性清空。

源码视角：通知的两个数据模型

从源码结构看，界面中的"收件箱/历史"能力由 NotificationMessage 模型支撑：它通过target_object/source_object两个 GenericForeignKey 记录通知的"目标对象"与"来源对象"，并把消息关联到具体的接收用户。

而防重复通知则依赖 NotificationEntry：

class NotificationEntry(MetaMixin): """A NotificationEntry records the last time a particular notification was sent out. It is recorded to ensure that notifications are not sent out "too often" to users. """ key = models.CharField(max_length=250, blank=False) # 如 'part.notify_low_stock' uid = models.CharField(max_length=255) # 触发实例 ID（支持 UUID 主键） class Meta: unique_together = [('key', 'uid')] @classmethod def check_recent(cls, key, uid, delta): """检查某类通知是否在指定时间窗口内已发送过""" since = InvenTree.helpers.current_date() - delta entries = cls.objects.filter(key=key, uid=uid, updated__gte=since) return entries.exists() @classmethod def notify(cls, key, uid): """记录一次已发送""" entry, _ = cls.objects.get_or_create(key=key, uid=uid) entry.save()

key + uid唯一约束加"最近更新时间"字段，构成了 InvenTree 通知去重（rate limiting）的基础设施——后文的低库存通知正是建立在这套机制之上。

低库存通知（Low Stock Notification）

触发规则

文档描述的规则是：当一个零件设置了"最低库存（minimum stock）"阈值后，若该零件的库存水平跌落到配置值以下，即可生成"低库存"通知；所有订阅了该零件通知的用户都会收到邮件。

在源码中，阈值对应 Part 模型的 minimum_stock 字段：

minimum_stock = models.DecimalField( max_digits=19, decimal_places=6, default=0, validators=[MinValueValidator(0)], verbose_name=_('Minimum Stock'), help_text=_('Minimum allowed stock level'), )

注意其默认值为0，且允许最多 6 位小数——这意味着"低库存通知"对未显式配置阈值的零件不会触发，小数精度的库存量（如按长度计量的线材）也能正确比较。

判定逻辑非常直接，见 Part.is_part_low_on_stock：

def is_part_low_on_stock(self): """Returns True if the total stock for this part is less than the minimum stock level.""" return self.get_stock_count() < self.minimum_stock

完整触发链路

低库存检查并非在每次库存变动时同步执行，而是一条信号 → 异步任务 → 分发的链路：

第 1 步：库存信号触发异步检查。在 stock/models.py 中，StockItem的post_save与post_delete信号都会把检查投递到后台任务队列：

@receiver(post_save, sender=StockItem, dispatch_uid='stock_item_post_save_log') def after_save_stock_item(sender, instance: StockItem, created, **kwargs): """Hook function to be executed after StockItem object is saved/updated.""" from part import tasks as part_tasks if InvenTree.ready.isImportingData() or InvenTree.ready.isRunningMigrations(): return if InvenTree.ready.canAppAccessDatabase(allow_test=True): InvenTree.tasks.offload_task( part_tasks.notify_low_stock_if_required, instance.part.pk, group='notification', # 归入 notification 任务组，避免阻塞主流程 force_async=True, )

post_delete信号（after_delete_stock_item）中有相同的调用。此外，Part 自身保存后（例如修改了minimum_stock阈值）也会调度同一条检查任务——因此"改阈值"和"改库存"两条路径都会触发预警评估。

第 2 步：后台任务逐层上溯判断阈值。notify_low_stock_if_required 是检查入口：

def notify_low_stock_if_required(part_id: int): """Check if the stock quantity has fallen below the minimum threshold of part.""" from part.models import Part try: part = Part.objects.get(pk=part_id) except Part.DoesNotExist: ... return # Run "up" the tree, to allow notification for "parent" parts parts = part.get_ancestors(include_self=True, ascending=True) for p in parts: if part.active and p.is_part_low_on_stock(): offload_task(notify_low_stock, p, group='notification')

这里的get_ancestors体现了一个文档未展开的细节：检查会沿版本链向上遍历，对包括自身在内的所有祖先零件逐一判断。也就是说，当某个变体零件的库存过低时，其模板零件（template part）若同样低于阈值，也会触发对应通知——这与后文"订阅模板零件可覆盖其变体"的订阅语义是配套的。

第 3 步：真正发送通知。notify_low_stock 组装邮件上下文并调用统一的分发入口：

def notify_low_stock(part: Model): """Notify interested users that a part is 'low stock'. Rules: - Triggered when the available stock for a given part falls be low the configured threshold - A notification is delivered to any users who are 'subscribed' to this part """ # Do not trigger low-stock notifications for inactive parts if not part.active: return 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_model.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 )

几个值得注意的实现细节：

• 停用零件不发通知：part.active为假时直接返回，避免对已归档零件发出无效预警；
• 通知类别名为part.notify_low_stock：这个字符串正是NotificationEntry.key的取值，也是去重窗口的依据；
• 邮件正文由独立模板email/low_stock_notification.html渲染，并附带指向该零件页面的绝对链接，方便收件人一键跳转；
• 接收人不是硬编码的，而是通过target_fnc=part.get_subscribers动态解析订阅者列表（详见订阅机制一节）。

去重：一天内同类通知只发一次

trigger_notification 在真正分发前会执行去重检查：

# Check if we have notified recently... delta = timedelta(days=1) if check_recent and common.models.NotificationEntry.check_recent( category, obj_ref_value, delta ): logger.info( "Notification '%s' has recently been sent for '%s' - SKIPPING", category, obj, ) return

即同一个零件（uid）的低库存通知（key='part.notify_low_stock'）在最近 24 小时内已经发送过就会跳过，并在日志中记录 "has recently been sent ... SKIPPING"。发送成功后再调用NotificationEntry.notify()落库刷新时间戳。这套机制保证了即使库存反复在阈值附近抖动，用户也不会被同一零件的重复邮件轰炸。

另外，trigger_notification 入口 还会在数据导入或重建期间（isImportingData()/isRebuildingData()）直接返回，避免批量导入触发风暴式通知。

生产工单通知（Build Order Notification）

文档描述的规则是：当创建一个新生产工单时，InvenTree 会检查完成该工单所需的零件是否有低库存的；若有，则向订阅了"被生产零件"的用户生成通知。

从源码结构看，这一场景实际由两个机制共同覆盖：

机制一：工单创建时通知相关方。Build 模型的 post_save 信号 中，新工单创建后会调用notify_responsible，并把被生产零件的全部订阅者作为额外接收人加入：

@receiver(post_save, sender=Build, dispatch_uid='build_post_save_log') def after_save_build(sender, instance: Build, created: bool, **kwargs): ... if created: # A new Build has just been created # Generate initial BuildLine objects for the Build instance.create_build_line_items() # Notify the responsible users that the build order has been created InvenTree.helpers_model.notify_responsible( instance, sender, exclude=instance.issued_by, extra_users=instance.part.get_subscribers(), )

extra_users=instance.part.get_subscribers()正是文档所说"通知订阅了被生产零件的用户"的代码体现：这些用户会收到"新工单已创建"类通知（其文案来自 InvenTreeNotificationBodies.NewOrder 定义的A new order has been created and assigned to you），并在其中获知该工单的物料准备情况。

机制二：工单消耗库存引发的低库存回检。当生产工单实际消耗库存时，bulk_update()批量写库不会触发StockItem的post_save信号，因此 Build 的库存消耗逻辑 会显式地为本次触碰到的每个零件补排一次低库存检查：

# bulk_update()/bulk_create() above do not fire StockItem's post_save signal, # which normally triggers a low-stock check for the affected part - so queue # that check explicitly, once per distinct part touched by this call touched_part_ids = {item.part_id for item in seen_stock_items.values()} InvenTree.tasks.bulk_offload_task( part.tasks.notify_low_stock_if_required, [((part_id,), {}) for part_id in touched_part_ids], group='notification', force_async=True, )

这解释了文档中"新建工单 → 检查所需零件是否低库存"的闭环：工单拉走库存后，受影响零件会立即进入与手动出入库相同的一条检查管线（notify_low_stock_if_required→is_part_low_on_stock→ 去重 → 分发），从而在产线消耗物料、库存跌破阈值的第一时间把预警推送给订阅者。

订阅通知（Subscribing to Notifications）

用户可以对Part（零件）或Part Category（零件分类）进行"订阅"，从而在相关事件发生时收到通知。

订阅一个零件

文档说明：订阅某个零件后，用户将收到以下范围事件的通知：

• 该零件本身；
• 该零件的所有变体（variant）。

在界面上，零件页面的订阅图标呈高亮状态表示已订阅，灰色表示未订阅；点击该图标即切换订阅状态。

源码层面，订阅状态由PartStar模型记录，解析逻辑见 Part.get_subscribers：

def get_subscribers(self, include_variants: bool = True, include_categories: bool = True): """Return a list of users who are 'subscribed' to this part. A user may 'subscribe' to this part in the following ways: a) Subscribing to the part instance directly b) Subscribing to a template part "above" this part (if it is a variant) c) Subscribing to the part category that this part belongs to d) Subscribing to a parent category of the category in c) """ subscribers = set() # Start by looking at direct subscriptions to a Part model queryset = PartStar.objects.all() if include_variants: queryset = queryset.filter(part__in=self.get_ancestors(include_self=True)) else: queryset = PartStar.filter(part=self) for star in queryset: subscribers.add(star.user) if include_categories and self.category: for sub in self.category.get_subscribers(): subscribers.add(sub) return list(subscribers)

对照文档可确认"变体"语义的实现方式：当include_variants=True时，查询会覆盖self.get_ancestors(include_self=True)——即该零件自身及其全部版本祖先。因此订阅一个模板零件（template part），其下所有变体零件的事件都会命中订阅者；订阅某个具体变体，则只覆盖该变体本身及其更上层模板的订阅集合中的直接订阅记录。

切换订阅状态的方法同样在模型层：Part.set_starred 在已订阅时删除对应PartStar记录，未订阅时创建记录。源码注释特别提到一点：取消对单个零件的订阅，并不会阻止用户经由父零件或分类订阅继续收到通知——这提示管理订阅时需要注意订阅来源的层级。

订阅一个零件分类

订阅分类后的通知范围（文档原文列举的四类）：

• 该分类本身；
• 其下的所有子分类（更低层级）；
• 该分类中包含的所有零件；
• 更低层级分类中包含的所有零件。

操作方式与零件订阅一致：点击分类界面上的通知图标即可切换。

分类订阅的解析见 PartCategory.get_subscribers：

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

PartCategoryStar保存的是"某用户星标了某分类"的关系。分类订阅之所以能覆盖"下级分类中的零件"，是因为零件级get_subscribers在解析时把self.category.get_subscribers()的结果并入（见上文 c/d 两种订阅方式）：只要零件所属分类的任意祖先链上有订阅记录，该订阅者就会被纳入。

通知分发管线：从订阅者到邮件

前面各节的所有事件最终都汇聚到 trigger_notification 这一个统一入口。从源码结构看，它的处理流程可以分为五段：

1. 上下文保护：数据导入/重建期间直接返回，不产生任何通知；
2. 对象引用解析：按obj_ref → pk → id → uid的顺序解析触发对象的引用值，解析失败抛出KeyError；
3. 去重检查：默认检查最近 24 小时内是否已发送过同类通知（check_recent=True），命中则记录日志并跳过；
4. 接收人解析与过滤：目标可以是User、Group（展开为该组所有用户）或Owner（InvenTree 的"负责人"抽象，可指向用户或用户组），随后过滤掉非活跃用户以及对触发对象无 view 权限的用户——这意味着订阅者若失去了对该零件的查看权限，也不会收到通知：

# Filter out any users who are inactive, or do not have the required model permissions valid_users = list( filter( lambda u: ( u and u.is_active and (not obj or check_user_permission(u, obj, 'view')) ), list(target_users), ) )

5. 插件化投递：遍历所有实现了NOTIFICATIONmixin 的已启用插件，调用plugin.filter_targets()（插件可进一步过滤接收人）和plugin.send_notification()完成实际投递（内置邮件插件即在此链路中工作）；若任一插件发送成功，则调用NotificationEntry.notify()落库记录，供后续去重使用。

这种"事件 → 订阅者解析 → 去重 → 插件投递"的分层设计，使得低库存、新工单、呆滞库存等不同事件可以共享同一套收件人语义与防骚扰机制，同时新通知渠道（如 Slack 类插件）只需实现通知插件接口即可接入。

小结与相关文件索引

InvenTree 的零件通知体系可以概括为：订阅（PartStar / PartCategoryStar）定义"谁该收到"，信号 + 后台任务定义"何时检查"，minimum_stock阈值定义"何时告警"，NotificationEntry定义"多久不重复告警"，插件体系负责"怎么送达"。对使用者而言，启用该功能的最小闭环是：配置邮件服务 → 在 Notifications 设置中启用 → 为关键零件设置minimum_stock→ 点击零件/分类上的订阅图标；此后库存跌破阈值或工单拉料导致缺料时，订阅者即可收到邮件与界面通知。

核心文件索引：

【免费下载链接】InvenTreeOpen Source Inventory Management System项目地址: https://gitcode.com/GitHub_Trending/in/InvenTree