Coverage for src/backend/InvenTree/part/tasks.py: 18%
202 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
1"""Background task definitions for the 'part' app."""
3from datetime import datetime, timedelta
4from typing import Optional
6from django.core.exceptions import ValidationError
7from django.db.models import Model
8from django.utils.translation import gettext_lazy as _
10import structlog
11from opentelemetry import trace
13import common.currency
14import common.notifications
15import InvenTree.helpers
16import InvenTree.helpers_model
17from common.settings import get_global_setting
18from InvenTree.tasks import (
19 ScheduledTask,
20 check_daily_holdoff,
21 offload_task,
22 record_task_success,
23 scheduled_task,
24)
26tracer = trace.get_tracer(__name__)
27logger = structlog.get_logger('inventree')
30@tracer.start_as_current_span('notify_low_stock')
31def notify_low_stock(part: Model):
32 """Notify interested users that a part is 'low stock'.
34 Rules:
35 - Triggered when the available stock for a given part falls be low the configured threshold
36 - A notification is delivered to any users who are 'subscribed' to this part
37 """
38 # Do not trigger low-stock notifications for inactive parts
39 if not part.active:
40 return
42 name = _('Low stock notification')
43 message = _(
44 f'The available stock for {part.name} has fallen below the configured minimum level'
45 )
46 context = {
47 'part': part,
48 'name': name,
49 'message': message,
50 'link': InvenTree.helpers_model.construct_absolute_url(part.get_absolute_url()),
51 'template': {'html': 'email/low_stock_notification.html', 'subject': name},
52 }
54 common.notifications.trigger_notification(
55 part, 'part.notify_low_stock', target_fnc=part.get_subscribers, context=context
56 )
59@tracer.start_as_current_span('notify_stale_stock')
60def notify_stale_stock(user, stale_items):
61 """Notify a user about all their stale stock items in one consolidated email.
63 Rules:
64 - Triggered when stock items' expiry dates are within the configured STOCK_STALE_DAYS
65 - One notification is delivered per user containing all their stale stock items
67 Arguments:
68 user: The user to notify
69 stale_items: List of stale stock items for this user
70 """
71 if not stale_items:
72 return
74 name = _('Stale stock notification')
75 item_count = len(stale_items)
77 if item_count == 1:
78 message = _('You have 1 stock item approaching its expiry date')
79 else:
80 message = _(f'You have {item_count} stock items approaching their expiry dates')
82 # Add absolute URLs and days until expiry for each stock item
83 stale_items_enhanced = []
84 today = InvenTree.helpers.current_date()
86 for stock_item in stale_items:
87 # Calculate days until expiry to print it clearly in table in email later
88 days_until_expiry = None
89 expiry_status = _('No expiry date')
91 if stock_item.expiry_date:
92 days_diff = (stock_item.expiry_date - today).days
94 if days_diff < 0:
95 days_until_expiry = days_diff # Keep negative value for template logic
96 expiry_status = _(f'Expired {abs(days_diff)} days ago')
97 elif days_diff == 0:
98 days_until_expiry = 0
99 expiry_status = _('Expires today')
100 else:
101 days_until_expiry = days_diff
102 expiry_status = _(f'{days_until_expiry} days')
104 item_data = {
105 'stock_item': stock_item,
106 'absolute_url': InvenTree.helpers_model.construct_absolute_url(
107 stock_item.get_absolute_url()
108 ),
109 'days_until_expiry': days_until_expiry,
110 'expiry_status': expiry_status,
111 }
112 stale_items_enhanced.append(item_data)
114 context = {
115 'stale_items': stale_items_enhanced,
116 'item_count': item_count,
117 'name': name,
118 'message': message,
119 'template': {'html': 'email/stale_stock_notification.html', 'subject': name},
120 }
122 # Use the first stock item as the trigger object for the notification system
123 trigger_object = stale_items[0] if stale_items else None
125 if trigger_object:
126 common.notifications.trigger_notification(
127 trigger_object,
128 'stock.notify_stale_stock',
129 targets=[user],
130 context=context,
131 check_recent=False,
132 )
135@tracer.start_as_current_span('notify_low_stock_if_required')
136def notify_low_stock_if_required(part_id: int):
137 """Check if the stock quantity has fallen below the minimum threshold of part.
139 If true, notify the users who have subscribed to the part
140 """
141 from part.models import Part
143 try:
144 part = Part.objects.get(pk=part_id)
145 except Part.DoesNotExist:
146 logger.warning(
147 'notify_low_stock_if_required: Part with ID %s does not exist', part_id
148 )
149 return
151 # Run "up" the tree, to allow notification for "parent" parts
152 parts = part.get_ancestors(include_self=True, ascending=True)
154 for p in parts:
155 if part.active and p.is_part_low_on_stock():
156 offload_task(notify_low_stock, p, group='notification')
159@tracer.start_as_current_span('check_stale_stock')
160@scheduled_task(ScheduledTask.DAILY)
161def check_stale_stock():
162 """Check all stock items for stale stock.
164 This function runs daily and checks if any stock items are approaching their expiry date
165 based on the STOCK_STALE_DAYS global setting.
167 For any stale stock items found, notifications are sent to users who have subscribed
168 to notifications for the respective parts. Each user receives one consolidated email
169 containing all their stale stock items.
170 """
171 from stock.models import StockItem
173 # Check if stock expiry functionality is enabled
174 if not get_global_setting('STOCK_ENABLE_EXPIRY', False, cache=False):
175 logger.info('Stock expiry functionality is not enabled - exiting')
176 return
178 # Check if STOCK_STALE_DAYS is configured
179 stale_days = int(get_global_setting('STOCK_STALE_DAYS', 0, cache=False))
181 if stale_days <= 0:
182 logger.info('Stock stale days is not configured or set to 0 - exiting')
183 return
185 today = InvenTree.helpers.current_date()
186 stale_threshold = today + timedelta(days=stale_days)
188 # Find stock items that are stale (expiry date within STOCK_STALE_DAYS)
189 stale_stock_items = StockItem.objects.filter(
190 StockItem.IN_STOCK_FILTER, # Only in-stock items
191 expiry_date__isnull=False, # Must have an expiry date
192 expiry_date__lt=stale_threshold, # Expiry date is within stale threshold
193 ).select_related('part', 'location') # Optimize queries
195 if not stale_stock_items.exists():
196 logger.info('No stale stock items found')
197 return
199 logger.info('Found %s stale stock items', stale_stock_items.count())
201 # Group stale stock items by user subscriptions
202 user_stale_items: dict[StockItem, list[StockItem]] = {}
204 for stock_item in stale_stock_items:
205 # Get all subscribers for this part
206 subscribers = stock_item.part.get_subscribers()
208 for user in subscribers:
209 if user not in user_stale_items:
210 user_stale_items[user] = []
211 user_stale_items[user].append(stock_item)
213 # Send one consolidated notification per user
214 for user, items in user_stale_items.items():
215 try:
216 offload_task(notify_stale_stock, user, items, group='notification')
217 except Exception as e:
218 logger.error(
219 'Error scheduling stale stock notification for user %s: %s',
220 user.username,
221 e,
222 )
224 logger.info(
225 'Scheduled stale stock notifications for %s users', len(user_stale_items)
226 )
229@tracer.start_as_current_span('update_part_pricing')
230def update_part_pricing(pricing: Model, counter: int = 0):
231 """Update cached pricing data for the specified PartPricing instance.
233 Arguments:
234 pricing: The target PartPricing instance to be updated
235 counter: How many times this function has been called in sequence
236 """
237 logger.info('Updating part pricing for %s', pricing.part)
239 pricing.update_pricing(
240 counter=counter,
241 previous_min=pricing.overall_min,
242 previous_max=pricing.overall_max,
243 )
246@tracer.start_as_current_span('check_missing_pricing')
247@scheduled_task(ScheduledTask.DAILY)
248def check_missing_pricing(limit=250):
249 """Check for parts with missing or outdated pricing information.
251 Tests for the following conditions:
252 - Pricing information does not exist
253 - Pricing information is "old"
254 - Pricing information is in the wrong currency
256 Arguments:
257 limit: Maximum number of parts to process at once
258 """
259 from part.models import Part, PartPricing
261 # Find any parts which have 'old' pricing information
262 days = int(get_global_setting('PRICING_UPDATE_DAYS', 30))
264 if days <= 0:
265 # Task does not run if the interval is zero
266 return
268 # Find parts for which pricing information has never been updated
269 results = PartPricing.objects.filter(updated=None)[:limit]
271 if results.count() > 0:
272 logger.info('Found %s parts with empty pricing', results.count())
274 for pp in results:
275 pp.schedule_for_update()
277 stale_date = datetime.now().date() - timedelta(days=days)
279 results = PartPricing.objects.filter(updated__lte=stale_date)[:limit]
281 if results.count() > 0:
282 logger.info('Found %s stale pricing entries', results.count())
284 for pp in results:
285 pp.schedule_for_update()
287 # Find any pricing data which is in the wrong currency
288 currency = common.currency.currency_code_default()
289 results = PartPricing.objects.exclude(currency=currency)
291 if results.count() > 0:
292 logger.info('Found %s pricing entries in the wrong currency', results.count())
294 for pp in results:
295 pp.schedule_for_update()
297 # Find any parts which do not have pricing information
298 results = Part.objects.filter(pricing_data=None)[:limit]
300 if results.count() > 0:
301 logger.info('Found %s parts without pricing', results.count())
303 for p in results:
304 pricing = p.pricing
305 pricing.save()
306 pricing.schedule_for_update()
309@tracer.start_as_current_span('scheduled_stocktake_reports')
310@scheduled_task(ScheduledTask.DAILY)
311def scheduled_stocktake_reports():
312 """Scheduled tasks for creating automated 'stocktake' entries.
314 A "stocktake" entry is a snapshot of the current stock levels for a given Part.
316 This task runs daily, and performs the following functions:
318 - Delete 'old' stocktake report files after the specified period
319 - Generate new reports at the specified period
320 """
321 import part.stocktake
322 from part.models import PartStocktake
324 if get_global_setting('STOCKTAKE_DELETE_OLD_ENTRIES', False, cache=False):
325 # First let's delete any old stock history entries
326 delete_n_days = int(
327 get_global_setting('STOCKTAKE_DELETE_DAYS', 365, cache=False)
328 )
330 threshold = datetime.now() - timedelta(days=delete_n_days)
331 old_entries = PartStocktake.objects.filter(date__lt=threshold)
333 if old_entries.exists():
334 logger.info('Deleting %s old stock entries', old_entries.count())
335 old_entries.delete()
337 # Next, check if stocktake functionality is enabled
338 if not get_global_setting('STOCKTAKE_ENABLE', False, cache=False):
339 logger.info('Stocktake functionality is not enabled - exiting')
340 return
342 report_n_days = int(get_global_setting('STOCKTAKE_AUTO_DAYS', 7, cache=False))
344 if report_n_days < 1:
345 logger.info('Stocktake auto reports are disabled, exiting')
346 return
348 if not check_daily_holdoff('STOCKTAKE_RECENT_REPORT', report_n_days):
349 logger.info('Stock history was recently generated - exiting')
350 return
352 # Generate new stock history entries
353 part.stocktake.perform_stocktake()
355 # Record the date of this task run
356 record_task_success('STOCKTAKE_RECENT_REPORT')
359@tracer.start_as_current_span('rebuild_supplier_parts')
360def rebuild_supplier_parts(part_id: int):
361 """Rebuild all SupplierPart objects for a given part.
363 This function is called when a bart part is changed,
364 which may cause the native units of any supplier parts to be updated
365 """
366 from company.models import SupplierPart
367 from part.models import Part
369 try:
370 prt = Part.objects.get(pk=part_id)
371 except Part.DoesNotExist:
372 return
374 supplier_parts = SupplierPart.objects.filter(part=prt)
376 n = supplier_parts.count()
378 for supplier_part in supplier_parts:
379 # Re-save the part, to ensure that the units have updated correctly
380 try:
381 supplier_part.full_clean()
382 supplier_part.save()
383 except ValidationError:
384 pass
386 if n > 0:
387 logger.info("Rebuilt %s supplier parts for part '%s'", n, prt.name)
390@tracer.start_as_current_span('check_bom_valid')
391def check_bom_valid(part_id: int):
392 """Recalculate the BOM checksum for all assemblies which include the specified Part.
394 Arguments:
395 part_id: The ID of the part for which to recalculate the BOM checksum.
396 """
397 from part.models import Part
399 try:
400 part = Part.objects.get(pk=part_id)
401 except Part.DoesNotExist:
402 logger.warning('check_bom_valid: Part with ID %s does not exist', part_id)
403 return
405 valid = part.is_bom_valid()
407 if valid != part.bom_validated:
408 part.bom_validated = valid
409 part.save()
412@tracer.start_as_current_span('validate_bom')
413def validate_bom(part_id: int, valid: bool, user_id: Optional[int] = None):
414 """Run BOM validation for the specified Part.
416 Arguments:
417 part_id: The ID of the part for which to validate the BOM.
418 valid: Boolean indicating whether the BOM is valid or not.
419 user_id: Optional ID of the user performing the validation.
420 """
421 from django.contrib.auth import get_user_model
423 from part.models import Part
425 User = get_user_model()
427 try:
428 part = Part.objects.get(pk=part_id)
429 except Part.DoesNotExist:
430 logger.warning('validate_bom: Part with ID %s does not exist', part_id)
431 return
433 if user_id: 433 ↛ 439line 433 didn't jump to line 439 because the condition on line 433 was always true
434 try:
435 user = User.objects.get(pk=user_id)
436 except User.DoesNotExist:
437 user = None
438 else:
439 user = None
441 part.validate_bom(user, valid=valid)