Coverage for cookbook/signals.py: 82%

1from functools import wraps 

2 

3from django.conf import settings 

4from django.contrib.auth.models import User 

5from django.contrib.postgres.search import SearchVector 

6from django.core.cache import caches 

7from django.db.models.signals import post_save 

8from django.dispatch import receiver 

9from django.utils import translation 

10from django_scopes import scope, scopes_disabled 

11 

12from cookbook.helper.cache_helper import CacheHelper 

13from cookbook.helper.shopping_helper import RecipeShoppingEditor 

14from cookbook.managers import DICTIONARY 

15from cookbook.models import (Food, MealPlan, PropertyType, Recipe, SearchFields, SearchPreference, 

16 Step, Unit, UserPreference) 

17 

18SQLITE = True 

19if settings.DATABASES['default']['ENGINE'] == 'django.db.backends.postgresql': 

20 SQLITE = False 

21 

22 

23# wraps a signal with the ability to set 'skip_signal' to avoid creating recursive signals 

24def skip_signal(signal_func): 

25 @wraps(signal_func) 

26 def _decorator(sender, instance, **kwargs): 

27 if not instance: 

28 return None 

29 if hasattr(instance, 'skip_signal'): 

30 return None 

31 return signal_func(sender, instance, **kwargs) 

32 

33 return _decorator 

34 

35 

36@receiver(post_save, sender=User) 

37def create_user_preference(sender, instance=None, created=False, **kwargs): 

38 if created: 

39 with scopes_disabled(): 

40 UserPreference.objects.get_or_create(user=instance) 

41 

42 

43@receiver(post_save, sender=SearchPreference) 

44def create_search_preference(sender, instance=None, created=False, **kwargs): 

45 if created: 

46 with scopes_disabled(): 

47 instance.unaccent.add(SearchFields.objects.get(name='Name')) 

48 instance.icontains.add(SearchFields.objects.get(name='Name')) 

49 instance.trigram.add(SearchFields.objects.get(name='Name')) 

50 

51 

52@receiver(post_save, sender=Recipe) 

53@skip_signal 

54def update_recipe_search_vector(sender, instance=None, created=False, **kwargs): 

55 if SQLITE: 

56 return 

57 language = DICTIONARY.get(translation.get_language(), 'simple') 

58 # these indexed fields are space wide, reading user preferences would lead to inconsistent behavior 

59 instance.name_search_vector = SearchVector('name__unaccent', weight='A', config=language) 

60 instance.desc_search_vector = SearchVector('description__unaccent', weight='C', config=language) 

61 try: 

62 instance.skip_signal = True 

63 instance.save() 

64 finally: 

65 del instance.skip_signal 

66 

67 

68@receiver(post_save, sender=Step) 

69@skip_signal 

70def update_step_search_vector(sender, instance=None, created=False, **kwargs): 

71 if SQLITE: 

72 return 

73 language = DICTIONARY.get(translation.get_language(), 'simple') 

74 instance.search_vector = SearchVector('instruction__unaccent', weight='B', config=language) 

75 try: 

76 instance.skip_signal = True 

77 instance.save() 

78 finally: 

79 del instance.skip_signal 

80 

81 

82@receiver(post_save, sender=Food) 

83@skip_signal 

84def update_food_inheritance(sender, instance=None, created=False, **kwargs): 

85 if not instance: 

86 return 

87 

88 inherit = instance.inherit_fields.all() 

89 # nothing to apply from parent and nothing to apply to children 

90 if (not instance.parent or inherit.count() == 0) and instance.numchild == 0: 

91 return 

92 

93 inherit = inherit.values_list('field', flat=True) 

94 # apply changes from parent to instance for each inherited field 

95 if instance.parent and inherit.count() > 0: 

96 parent = instance.get_parent() 

97 for field in ['ignore_shopping', 'substitute_children', 'substitute_siblings']: 

98 if field in inherit: 

99 setattr(instance, field, getattr(parent, field, None)) 

100 # if supermarket_category is not set, do not cascade - if this becomes non-intuitive can change 

101 if 'supermarket_category' in inherit and parent.supermarket_category: 

102 instance.supermarket_category = parent.supermarket_category 

103 try: 

104 instance.skip_signal = True 

105 instance.save() 

106 finally: 

107 del instance.skip_signal 

108 

109 # apply changes to direct children - depend on save signals for those objects to cascade inheritance down 

110 for child in instance.get_children().filter(inherit_fields__in=Food.inheritable_fields): 

111 # set inherited field values 

112 for field in (inherit_fields := ['ignore_shopping', 'substitute_children', 'substitute_siblings']): 

113 if field in instance.inherit_fields.values_list('field', flat=True): 

114 setattr(child, field, getattr(instance, field, None)) 

115 

116 # don't cascade empty supermarket category 

117 if instance.supermarket_category and 'supermarket_category' in inherit_fields: 

118 setattr(child, 'supermarket_category', getattr(instance, 'supermarket_category', None)) 

119 

120 child.save() 

121 

122 

123@receiver(post_save, sender=MealPlan) 

124def auto_add_shopping(sender, instance=None, created=False, weak=False, **kwargs): 

125 print("MEAL_AUTO_ADD Signal trying to auto add to shopping") 

126 if not instance: 

127 print("MEAL_AUTO_ADD Instance is none") 

128 return 

129 

130 try: 

131 space = instance.get_space() 

132 user = instance.get_owner() 

133 with scope(space=space): 

134 slr_exists = instance.shoppinglistrecipe_set.exists() 

135 

136 if not created and slr_exists: 

137 for x in instance.shoppinglistrecipe_set.all(): 

138 # assuming that permissions checks for the MealPlan have happened upstream 

139 if instance.servings != x.servings: 

140 SLR = RecipeShoppingEditor(id=x.id, user=user, space=instance.space) 

141 SLR.edit_servings(servings=instance.servings) 

142 elif not user.userpreference.mealplan_autoadd_shopping or not instance.recipe: 

143 print("MEAL_AUTO_ADD No recipe or no setting") 

144 return 

145 

146 if created: 

147 SLR = RecipeShoppingEditor(user=user, space=space) 

148 SLR.create(mealplan=instance, servings=instance.servings) 

149 print("MEAL_AUTO_ADD Created SLR") 

150 except AttributeError: 

151 pass 

152 

153 

154@receiver(post_save, sender=Unit) 

155def clear_unit_cache(sender, instance=None, created=False, **kwargs): 

156 if instance: 

157 caches['default'].delete(CacheHelper(instance.space).BASE_UNITS_CACHE_KEY) 

158 

159 

160@receiver(post_save, sender=PropertyType) 

161def clear_property_type_cache(sender, instance=None, created=False, **kwargs): 

162 if instance: 

163 caches['default'].delete(CacheHelper(instance.space).PROPERTY_TYPE_CACHE_KEY)