Coverage for src/common/views.py: 65%

1""" 

2Views and async utilities for common backend operations, including: 

3- Sending reminder emails to users with uncompleted shopping bags 

4- Providing reviewer-only API to inspect old shopping bag items 

5""" 

6 

7from django.conf import settings 

8from django.contrib.auth import get_user_model 

9from asgiref.sync import sync_to_async 

10import asyncio 

11 

12from django.template.loader import render_to_string 

13from django.utils.html import strip_tags 

14from django.utils import timezone 

15from datetime import timedelta 

16from src.shopping_bags.models import ShoppingBag 

17from django.http import JsonResponse 

18from rest_framework.views import APIView 

19from rest_framework.response import Response 

20from rest_framework.permissions import AllowAny 

21 

22from drf_spectacular.utils import extend_schema 

23 

24from src.common.tasks import _send_email 

25from src.common.permissions import IsOrderManager 

26 

27UserModel = get_user_model() 

28 

29 

30async def send_email( 

31 subject, message, html_message, from_email, recipient_list 

32): 

33 """ 

34 Asynchronously send an email using the configured email backend. 

35 Uses a Celery task to send the email in the background. 

36 """ 

37 _send_email.delay( 

38 subject, message, html_message, from_email, recipient_list 

39 ) 

40 

41 

42async def notify_users_they_have_uncompleted_orders(request): 

43 """ 

44 Finds users with shopping bags older than one day and sends them a reminder email. 

45 - Queries ShoppingBag for bags created more than one day ago. 

46 - Sends an email to each user with such a bag. 

47 - Returns a JsonResponse with the list of user IDs notified. 

48 """ 

49 one_day_ago = timezone.now() - timedelta(days=1) 

50 

51 user_ids = await sync_to_async( 

52 lambda: list( 

53 ShoppingBag.objects.filter( 

54 created_at__lt=one_day_ago, 

55 ) 

56 .values_list( 

57 'user', 

58 flat=True, 

59 ) 

60 .distinct() 

61 ) 

62 )() 

63 

64 users = await sync_to_async(list)( 

65 UserModel.objects.filter( 

66 id__in=user_ids, 

67 ) 

68 ) 

69 

70 html_message = render_to_string('mailer/uncompleted-orders.html') 

71 plain_message = strip_tags(html_message) 

72 

73 email_tasks = [ 

74 send_email( 

75 subject='Don’t Forget the Items in Your Shopping Bag!', 

76 message=plain_message, 

77 html_message=html_message, 

78 from_email=settings.EMAIL_HOST_USER, 

79 recipient_list=(user.email,), 

80 ) 

81 for user in users 

82 ] 

83 

84 await asyncio.gather(*email_tasks) 

85 

86 return JsonResponse({'detail': 'Emails have been sent'}) 

87 

88 

89class ShoppingBagReminderInfoView(APIView): 

90 """ 

91 API endpoint for reviewers to inspect shopping bags older than one day. 

92 - Only accessible to users with order manager permissions. 

93 - Returns a list of bag items with user and product info. 

94 """ 

95 

96 permission_classes = [IsOrderManager] 

97 

98 @extend_schema( 

99 summary='Get shopping bag reminders', 

100 description='Returns shopping bags older than one day for order managers to review' 

101 ) 

102 def get(self, request): 

103 one_day_ago = timezone.now() - timedelta(days=1) 

104 bag_items = ShoppingBag.objects.filter( 

105 created_at__lt=one_day_ago 

106 ).select_related( 

107 'user', 

108 'inventory', 

109 ) 

110 

111 data = [ 

112 { 

113 'id': item.id, 

114 'created_at': item.created_at, 

115 'quantity': item.quantity, 

116 'total_price': ( 

117 item.inventory.price * item.quantity 

118 if hasattr(item.inventory, 'price') 

119 else None 

120 ), 

121 'user_email': item.user.email if item.user else None, 

122 'product_info': f'Size: {item.inventory.size} / Price: {item.inventory.price}', 

123 } 

124 for item in bag_items 

125 ] 

126 

127 return Response(data) 

128 

129 

130class WakeUpServerView(APIView): 

131 permission_classes = [AllowAny] 

132 

133 def get(self, request): 

134 return JsonResponse({'detail': 'The server was awakened.'})