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

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 

11from django.template.loader import render_to_string 

12from django.utils.html import strip_tags 

13from django.utils import timezone 

14from datetime import timedelta 

15from src.shopping_bags.models import ShoppingBag 

16from django.http import JsonResponse 

17from rest_framework.views import APIView 

18from rest_framework.response import Response 

19 

20from src.common.tasks import _send_email 

21from src.common.permissions import IsOrderManager 

22 

23UserModel = get_user_model() 

24 

25 

26async def send_email( 

27 subject, message, html_message, from_email, recipient_list 

28): 

29 """ 

30 Asynchronously send an email using the configured email backend. 

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

32 """ 

33 _send_email.delay( 

34 subject, message, html_message, from_email, recipient_list 

35 ) 

36 

37 

38async def notify_users_they_have_uncompleted_orders(request): 

39 """ 

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

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

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

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

44 """ 

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

46 

47 user_ids = await sync_to_async( 

48 lambda: list( 

49 ShoppingBag.objects.filter( 

50 created_at__lt=one_day_ago, 

51 ) 

52 .values_list( 

53 'user', 

54 flat=True, 

55 ) 

56 .distinct() 

57 ) 

58 )() 

59 

60 users = await sync_to_async(list)( 

61 UserModel.objects.filter( 

62 id__in=user_ids, 

63 ) 

64 ) 

65 

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

67 plain_message = strip_tags(html_message) 

68 

69 email_tasks = [ 

70 send_email( 

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

72 message=plain_message, 

73 html_message=html_message, 

74 from_email=settings.EMAIL_HOST_USER, 

75 recipient_list=(user.email,), 

76 ) 

77 for user in users 

78 ] 

79 

80 await asyncio.gather(*email_tasks) 

81 

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

83 

84 

85class ShoppingBagReminderInfoView(APIView): 

86 """ 

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

88 - Only accessible to users with order manager permissions. 

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

90 """ 

91 

92 permission_classes = [IsOrderManager] 

93 

94 def get(self, request): 

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

96 bag_items = ShoppingBag.objects.filter( 

97 created_at__lt=one_day_ago 

98 ).select_related( 

99 'user', 

100 'inventory', 

101 ) 

102 

103 data = [ 

104 { 

105 'id': item.id, 

106 'created_at': item.created_at, 

107 'quantity': item.quantity, 

108 'total_price': ( 

109 item.inventory.price * item.quantity 

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

111 else None 

112 ), 

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

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

115 } 

116 for item in bag_items 

117 ] 

118 

119 return Response(data)