Coverage for src/backend/InvenTree/report/tasks.py: 15%

1"""Background tasks for the report app.""" 

2 

3from django.contrib.auth import get_user_model 

4 

5import structlog 

6from opentelemetry import trace 

7 

8from InvenTree.exceptions import log_error 

9 

10tracer = trace.get_tracer(__name__) 

11logger = structlog.get_logger('inventree') 

12 

13 

14@tracer.start_as_current_span('print_reports') 

15def print_reports( 

16 template_id: int, item_ids: list[int], output_id: int, user_id: int, **kwargs 

17): 

18 """Print multiple reports against the provided template. 

19 

20 Arguments: 

21 template_id: The ID of the ReportTemplate to use 

22 item_ids: List of item IDs to generate the report against 

23 output_id: The ID of the DataOutput to use 

24 user_id: The ID of the user to associate with the generated report 

25 

26 This function is intended to be called by the background worker, 

27 and will continuously update the status of the DataOutput object. 

28 """ 

29 from common.models import DataOutput 

30 from report.models import ReportTemplate 

31 

32 try: 

33 template = ReportTemplate.objects.get(pk=template_id) 

34 output = DataOutput.objects.get(pk=output_id) 

35 except Exception: 

36 log_error('report.tasks.print_reports') 

37 return 

38 

39 # Fetch user information 

40 user = None 

41 

42 if user_id: 

43 try: 

44 user = get_user_model().objects.get(pk=user_id) 

45 except Exception: 

46 log_error('report.tasks.print_reports', user_id=user_id) 

47 

48 if not user: 

49 user = getattr(output, 'user', None) 

50 

51 # Fetch the items to be included in the report 

52 model = template.get_model() 

53 items = model.objects.filter(pk__in=item_ids) 

54 

55 # Ensure they are sorted by the order of the provided item IDs 

56 items = sorted(items, key=lambda item: item_ids.index(item.pk)) 

57 

58 template.print(items, output=output, user=user) 

59 

60 

61@tracer.start_as_current_span('print_labels') 

62def print_labels( 

63 template_id: int, 

64 item_ids: list[int], 

65 output_id: int, 

66 user_id: int, 

67 plugin_slug: str, 

68 **kwargs, 

69): 

70 """Print multiple labels against the provided template. 

71 

72 Arguments: 

73 template_id: The ID of the LabelTemplate to use 

74 item_ids: List of item IDs to generate the labels against 

75 output_id: The ID of the DataOutput to use 

76 user_id: The ID of the user to associate with the generated labels 

77 plugin_slug: The ID of the LabelPlugin to use 

78 

79 This function is intended to be called by the background worker, 

80 and will continuously update the status of the DataOutput object. 

81 """ 

82 from common.models import DataOutput 

83 from plugin.registry import registry 

84 from report.models import LabelTemplate 

85 

86 try: 

87 template = LabelTemplate.objects.get(pk=template_id) 

88 output = DataOutput.objects.get(pk=output_id) 

89 except Exception: 

90 log_error('report.tasks.print_labels') 

91 return 

92 

93 # Fetch user information 

94 user = None 

95 

96 if user_id: 

97 try: 

98 user = get_user_model().objects.get(pk=user_id) 

99 except Exception: 

100 log_error('report.tasks.print_labels', user_id=user_id) 

101 

102 if not user: 

103 user = getattr(output, 'user', None) 

104 

105 # Fetch the items to be included in the report 

106 model = template.get_model() 

107 items = model.objects.filter(pk__in=item_ids) 

108 

109 # Ensure they are sorted by the order of the provided item IDs 

110 items = sorted(items, key=lambda item: item_ids.index(item.pk)) 

111 

112 plugin = registry.get_plugin(plugin_slug, active=True) 

113 

114 if not plugin: 

115 logger.warning("Label printing plugin '%s' not found", plugin_slug) 

116 return 

117 

118 # Extract optional arguments for label printing 

119 options = kwargs.pop('options') or {} 

120 

121 template.print(items, plugin, output=output, user=user, options=options)