619 lines
24 KiB
Python
619 lines
24 KiB
Python
"""
|
|
Views for the notifications API.
|
|
"""
|
|
from datetime import datetime, timedelta
|
|
|
|
from django.conf import settings
|
|
from django.db.models import Count
|
|
from django_ratelimit.core import is_ratelimited
|
|
from django.shortcuts import get_object_or_404
|
|
from django.utils.translation import gettext as _
|
|
from zoneinfo import ZoneInfo
|
|
from rest_framework import generics, status
|
|
from rest_framework.decorators import api_view
|
|
from rest_framework.generics import UpdateAPIView
|
|
from rest_framework.response import Response
|
|
from rest_framework.views import APIView
|
|
|
|
from openedx.core.djangoapps.notifications.email.utils import update_user_preferences_from_patch, username_from_hash
|
|
from openedx.core.djangoapps.notifications.models import NotificationPreference
|
|
from openedx.core.djangoapps.notifications.permissions import allow_any_authenticated_user
|
|
|
|
from .base_notification import COURSE_NOTIFICATION_APPS, NotificationAppManager, COURSE_NOTIFICATION_TYPES, \
|
|
NotificationTypeManager, filter_notification_types_by_app
|
|
from .events import (
|
|
notification_preference_update_event,
|
|
notification_read_event,
|
|
notification_tray_opened_event,
|
|
notifications_app_all_read_event
|
|
)
|
|
from .models import Notification
|
|
from .serializers import (
|
|
NotificationSerializer,
|
|
UserNotificationPreferenceUpdateAllSerializer,
|
|
add_info_to_notification_config,
|
|
add_non_editable_in_preference
|
|
)
|
|
from .tasks import create_notification_preference
|
|
from .utils import (
|
|
get_show_notifications_tray,
|
|
exclude_inaccessible_preferences, create_account_notification_pref_if_not_exists
|
|
)
|
|
|
|
|
|
@allow_any_authenticated_user()
|
|
class NotificationListAPIView(generics.ListAPIView):
|
|
"""
|
|
API view for listing notifications for a user.
|
|
|
|
**Permissions**: User must be authenticated.
|
|
**Response Format** (paginated):
|
|
|
|
{
|
|
"results" : [
|
|
{
|
|
"id": (int) notification_id,
|
|
"app_name": (str) app_name,
|
|
"notification_type": (str) notification_type,
|
|
"content": (str) content,
|
|
"content_context": (dict) content_context,
|
|
"content_url": (str) content_url,
|
|
"last_read": (datetime) last_read,
|
|
"last_seen": (datetime) last_seen
|
|
},
|
|
...
|
|
],
|
|
"count": (int) total_number_of_notifications,
|
|
"next": (str) url_to_next_page_of_notifications,
|
|
"previous": (str) url_to_previous_page_of_notifications,
|
|
"page_size": (int) number_of_notifications_per_page,
|
|
|
|
}
|
|
|
|
Response Error Codes:
|
|
- 403: The requester cannot access resource.
|
|
"""
|
|
|
|
serializer_class = NotificationSerializer
|
|
|
|
def get_queryset(self):
|
|
"""
|
|
Override the get_queryset method to filter the queryset by app name, request.user and created
|
|
"""
|
|
expiry_date = datetime.now(ZoneInfo("UTC")) - timedelta(days=settings.NOTIFICATIONS_EXPIRY)
|
|
app_name = self.request.query_params.get('app_name')
|
|
|
|
if self.request.query_params.get('tray_opened'):
|
|
unseen_count = Notification.objects.filter(user_id=self.request.user, last_seen__isnull=True).count()
|
|
notification_tray_opened_event(self.request.user, unseen_count)
|
|
params = {
|
|
'user': self.request.user,
|
|
'created__gte': expiry_date,
|
|
'web': True
|
|
}
|
|
|
|
if app_name:
|
|
params['app_name'] = app_name
|
|
return Notification.objects.filter(**params).order_by('-created')
|
|
|
|
|
|
@allow_any_authenticated_user()
|
|
class NotificationCountView(APIView):
|
|
"""
|
|
API view for getting the unseen notifications count and show_notification_tray flag for a user.
|
|
"""
|
|
|
|
def get(self, request):
|
|
"""
|
|
Get the unseen notifications count and show_notification_tray flag for a user.
|
|
|
|
**Permissions**: User must be authenticated.
|
|
**Response Format**:
|
|
```json
|
|
{
|
|
"show_notifications_tray": (bool) show_notifications_tray,
|
|
"count": (int) total_number_of_unseen_notifications,
|
|
"count_by_app_name": {
|
|
(str) app_name: (int) number_of_unseen_notifications,
|
|
...
|
|
},
|
|
"notification_expiry_days": 60
|
|
}
|
|
```
|
|
**Response Error Codes**:
|
|
- 403: The requester cannot access resource.
|
|
"""
|
|
# Get the unseen notifications count for each app name.
|
|
count_by_app_name = (
|
|
Notification.objects
|
|
.filter(user_id=request.user, last_seen__isnull=True, web=True)
|
|
.values('app_name')
|
|
.annotate(count=Count('*'))
|
|
)
|
|
count_total = 0
|
|
show_notifications_tray = get_show_notifications_tray()
|
|
count_by_app_name_dict = {
|
|
app_name: 0
|
|
for app_name in COURSE_NOTIFICATION_APPS
|
|
}
|
|
|
|
for item in count_by_app_name:
|
|
app_name = item['app_name']
|
|
count = item['count']
|
|
count_total += count
|
|
count_by_app_name_dict[app_name] = count
|
|
|
|
return Response({
|
|
"show_notifications_tray": show_notifications_tray,
|
|
"count": count_total,
|
|
"count_by_app_name": count_by_app_name_dict,
|
|
"notification_expiry_days": settings.NOTIFICATIONS_EXPIRY,
|
|
})
|
|
|
|
|
|
@allow_any_authenticated_user()
|
|
class MarkNotificationsSeenAPIView(UpdateAPIView):
|
|
"""
|
|
API view for marking user's all notifications seen for a provided app_name.
|
|
"""
|
|
|
|
def update(self, request, *args, **kwargs):
|
|
"""
|
|
Marks all notifications for the given app name seen for the authenticated user.
|
|
|
|
**Args:**
|
|
app_name: The name of the app to mark notifications seen for.
|
|
**Response Format:**
|
|
A `Response` object with a 200 OK status code if the notifications were successfully marked seen.
|
|
**Response Error Codes**:
|
|
- 400: Bad Request status code if the app name is invalid.
|
|
"""
|
|
app_name = self.kwargs.get('app_name')
|
|
|
|
if not app_name:
|
|
return Response({'error': _('Invalid app name.')}, status=400)
|
|
|
|
notifications = Notification.objects.filter(
|
|
user=request.user,
|
|
app_name=app_name,
|
|
last_seen__isnull=True,
|
|
)
|
|
|
|
notifications.update(last_seen=datetime.now())
|
|
|
|
return Response({'message': _('Notifications marked as seen.')}, status=200)
|
|
|
|
|
|
@allow_any_authenticated_user()
|
|
class NotificationReadAPIView(APIView):
|
|
"""
|
|
API view for marking user notifications as read, either all notifications or a single notification
|
|
"""
|
|
|
|
def patch(self, request, *args, **kwargs):
|
|
"""
|
|
Marks all notifications or single notification read for the given
|
|
app name or notification id for the authenticated user.
|
|
|
|
Requests:
|
|
PATCH /api/notifications/read/
|
|
|
|
Parameters:
|
|
request (Request): The request object containing the app name or notification id.
|
|
{
|
|
"app_name": (str) app_name,
|
|
"notification_id": (int) notification_id
|
|
}
|
|
|
|
Returns:
|
|
- 200: OK status code if the notification or notifications were successfully marked read.
|
|
- 400: Bad Request status code if the app name is invalid.
|
|
- 403: Forbidden status code if the user is not authenticated.
|
|
- 404: Not Found status code if the notification was not found.
|
|
"""
|
|
notification_id = request.data.get('notification_id', None)
|
|
read_at = datetime.now(ZoneInfo("UTC"))
|
|
|
|
if notification_id:
|
|
notification = get_object_or_404(Notification, pk=notification_id, user=request.user)
|
|
first_time_read = notification.last_read is None
|
|
notification.last_read = read_at
|
|
notification.save()
|
|
notification_read_event(request.user, notification, first_time_read)
|
|
return Response({'message': _('Notification marked read.')}, status=status.HTTP_200_OK)
|
|
|
|
app_name = request.data.get('app_name', '')
|
|
|
|
if app_name in COURSE_NOTIFICATION_APPS:
|
|
notifications = Notification.objects.filter(
|
|
user=request.user,
|
|
app_name=app_name,
|
|
last_read__isnull=True,
|
|
)
|
|
notifications.update(last_read=read_at)
|
|
notifications_app_all_read_event(request.user, app_name)
|
|
return Response({'message': _('Notifications marked read.')}, status=status.HTTP_200_OK)
|
|
|
|
return Response({'error': _('Invalid app_name or notification_id.')}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
|
|
@api_view(['GET', 'POST'])
|
|
def preference_update_from_encrypted_username_view(request, username, patch=""):
|
|
"""
|
|
View to update user preferences from encrypted username and patch.
|
|
username and patch must be string
|
|
"""
|
|
if is_ratelimited(
|
|
request=request, group="unsubscribe", key=username_from_hash,
|
|
rate=settings.ONE_CLICK_UNSUBSCRIBE_RATE_LIMIT, increment=True,
|
|
):
|
|
return Response({"error": "Too many requests"}, status=status.HTTP_429_TOO_MANY_REQUESTS)
|
|
update_user_preferences_from_patch(username)
|
|
return Response({"result": "success"}, status=status.HTTP_200_OK)
|
|
|
|
|
|
@allow_any_authenticated_user()
|
|
class NotificationPreferencesView(APIView):
|
|
"""
|
|
API view to retrieve and structure the notification preferences for the
|
|
authenticated user.
|
|
"""
|
|
|
|
def get(self, request):
|
|
"""
|
|
Handles GET requests to retrieve notification preferences.
|
|
|
|
This method fetches the user's active notification preferences and
|
|
merges them with a default structure provided by NotificationAppManager.
|
|
This provides a complete view of all possible notifications and the
|
|
user's current settings for them.
|
|
|
|
Returns:
|
|
Response: A DRF Response object containing the structured
|
|
notification preferences or an error message.
|
|
"""
|
|
user_preferences_qs = NotificationPreference.objects.filter(user=request.user)
|
|
user_preferences_map = {pref.type: pref for pref in user_preferences_qs}
|
|
|
|
# Ensure all notification types are present in the user's preferences.
|
|
# If any are missing, create them with default values.
|
|
diff = set(COURSE_NOTIFICATION_TYPES.keys()) - set(user_preferences_map.keys())
|
|
missing_types = []
|
|
for missing_type in diff:
|
|
new_pref = create_notification_preference(
|
|
user_id=request.user.id,
|
|
notification_type=missing_type,
|
|
|
|
)
|
|
missing_types.append(new_pref)
|
|
user_preferences_map[missing_type] = new_pref
|
|
if missing_types:
|
|
NotificationPreference.objects.bulk_create(missing_types)
|
|
|
|
# If no user preferences are found, return an error response.
|
|
if not user_preferences_map:
|
|
return Response({
|
|
'status': 'error',
|
|
'message': 'No active notification preferences found for this user.'
|
|
}, status=status.HTTP_404_NOT_FOUND)
|
|
|
|
# Get the structured preferences from the NotificationAppManager.
|
|
# This will include all apps and their notification types.
|
|
structured_preferences = NotificationAppManager().get_notification_app_preferences()
|
|
|
|
for app_name, app_settings in structured_preferences.items():
|
|
notification_types = app_settings.get('notification_types', {})
|
|
|
|
# Process all notification types (core and non-core) in a single loop.
|
|
for type_name, type_details in notification_types.items():
|
|
if type_name == 'core':
|
|
if structured_preferences[app_name]['core_notification_types']:
|
|
# If the app has core notification types, use the first one as the type name.
|
|
# This assumes that the first core notification type is representative of the core settings.
|
|
notification_type = structured_preferences[app_name]['core_notification_types'][0]
|
|
else:
|
|
notification_type = 'core'
|
|
user_pref = user_preferences_map.get(notification_type)
|
|
else:
|
|
user_pref = user_preferences_map.get(type_name)
|
|
if user_pref:
|
|
# If a preference exists, update the dictionary for this type.
|
|
# This directly modifies the 'type_details' dictionary.
|
|
type_details['web'] = user_pref.web
|
|
type_details['email'] = user_pref.email
|
|
type_details['push'] = user_pref.push
|
|
type_details['email_cadence'] = user_pref.email_cadence
|
|
exclude_inaccessible_preferences(structured_preferences, request.user)
|
|
structured_preferences = add_non_editable_in_preference(
|
|
add_info_to_notification_config(structured_preferences)
|
|
)
|
|
return Response({
|
|
'status': 'success',
|
|
'message': 'Notification preferences retrieved successfully.',
|
|
'show_preferences': get_show_notifications_tray(),
|
|
'data': structured_preferences
|
|
}, status=status.HTTP_200_OK)
|
|
|
|
def put(self, request):
|
|
"""
|
|
Handles PUT requests to update notification preferences.
|
|
|
|
This method updates the user's notification preferences based on the
|
|
provided data in the request body. It expects a dictionary with
|
|
notification types and their settings.
|
|
|
|
Returns:
|
|
Response: A DRF Response object indicating success or failure.
|
|
"""
|
|
# Validate incoming data
|
|
serializer = UserNotificationPreferenceUpdateAllSerializer(data=request.data)
|
|
if not serializer.is_valid():
|
|
return Response({
|
|
'status': 'error',
|
|
'message': serializer.errors
|
|
}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
# Get validated data for easier access
|
|
validated_data = serializer.validated_data
|
|
|
|
# Build query set based on notification type
|
|
query_set = NotificationPreference.objects.filter(user_id=request.user.id)
|
|
|
|
if validated_data['notification_type'] == 'core':
|
|
# Get core notification types for the app
|
|
__, core_types = NotificationTypeManager().get_notification_app_preference(
|
|
notification_app=validated_data['notification_app']
|
|
)
|
|
query_set = query_set.filter(type__in=core_types)
|
|
else:
|
|
# Filter by single notification type
|
|
query_set = query_set.filter(type=validated_data['notification_type'])
|
|
|
|
# Prepare update data based on channel type
|
|
updated_data = self._prepare_update_data(validated_data)
|
|
|
|
# Update preferences
|
|
query_set.update(**updated_data)
|
|
|
|
# Log the event
|
|
self._log_preference_update_event(request.user, validated_data)
|
|
|
|
# Prepare and return response
|
|
response_data = self._prepare_response_data(validated_data)
|
|
return Response(response_data, status=status.HTTP_200_OK)
|
|
|
|
def _prepare_update_data(self, validated_data):
|
|
"""
|
|
Prepare the data dictionary for updating notification preferences.
|
|
|
|
Args:
|
|
validated_data (dict): Validated serializer data
|
|
|
|
Returns:
|
|
dict: Dictionary with update data
|
|
"""
|
|
channel = validated_data['notification_channel']
|
|
|
|
if channel == 'email_cadence':
|
|
return {channel: validated_data['email_cadence']}
|
|
else:
|
|
return {channel: validated_data['value']}
|
|
|
|
def _log_preference_update_event(self, user, validated_data):
|
|
"""
|
|
Log the notification preference update event.
|
|
|
|
Args:
|
|
user: The user making the update
|
|
validated_data (dict): Validated serializer data
|
|
"""
|
|
event_data = {
|
|
'notification_app': validated_data['notification_app'],
|
|
'notification_type': validated_data['notification_type'],
|
|
'notification_channel': validated_data['notification_channel'],
|
|
'value': validated_data.get('value'),
|
|
'email_cadence': validated_data.get('email_cadence'),
|
|
}
|
|
notification_preference_update_event(user, [], event_data)
|
|
|
|
def _prepare_response_data(self, validated_data):
|
|
"""
|
|
Prepare the response data dictionary.
|
|
|
|
Args:
|
|
validated_data (dict): Validated serializer data
|
|
|
|
Returns:
|
|
dict: Response data dictionary
|
|
"""
|
|
email_cadence = validated_data.get('email_cadence', None)
|
|
# Determine the updated value
|
|
updated_value = validated_data.get('value', email_cadence if email_cadence else None)
|
|
|
|
# Determine the channel
|
|
channel = validated_data.get('notification_channel')
|
|
if not channel and validated_data.get('email_cadence'):
|
|
channel = 'email_cadence'
|
|
|
|
return {
|
|
'status': 'success',
|
|
'message': 'Notification preferences update completed',
|
|
'show_preferences': get_show_notifications_tray(),
|
|
'data': {
|
|
'updated_value': updated_value,
|
|
'notification_type': validated_data['notification_type'],
|
|
'channel': channel,
|
|
'app': validated_data['notification_app'],
|
|
}
|
|
}
|
|
|
|
|
|
@allow_any_authenticated_user()
|
|
class NotificationPreferencesViewV3(APIView):
|
|
"""
|
|
API view to retrieve and structure the notification preferences for the
|
|
authenticated user.
|
|
"""
|
|
|
|
def get(self, request):
|
|
"""
|
|
Handles GET requests to retrieve notification preferences.
|
|
|
|
This method fetches the user's active notification preferences and
|
|
merges them with a default structure provided by NotificationAppManager.
|
|
This provides a complete view of all possible notifications and the
|
|
user's current settings for them.
|
|
|
|
Returns:
|
|
Response: A DRF Response object containing the structured
|
|
notification preferences or an error message.
|
|
"""
|
|
user_preferences_qs = NotificationPreference.objects.filter(user=request.user)
|
|
|
|
# Ensure all notification types are present in the user's preferences.
|
|
user_preferences_qs = create_account_notification_pref_if_not_exists(
|
|
user_ids=[request.user.id],
|
|
existing_preferences=user_preferences_qs,
|
|
notification_types=COURSE_NOTIFICATION_TYPES.keys()
|
|
)
|
|
structured_preferences = {
|
|
app_name: {
|
|
'notification_types': {},
|
|
'enabled': COURSE_NOTIFICATION_APPS[app_name].get('enabled', True),
|
|
'non_editable': []
|
|
|
|
} for app_name in COURSE_NOTIFICATION_APPS.keys()}
|
|
|
|
for user_preference in user_preferences_qs:
|
|
app_name = user_preference.app
|
|
type_name = user_preference.type
|
|
|
|
if user_preference.is_grouped:
|
|
structured_preferences[app_name]['notification_types']['grouped_notification'] = {
|
|
**user_preference.config
|
|
}
|
|
continue
|
|
|
|
structured_preferences[app_name]['notification_types'][type_name] = {**user_preference.config}
|
|
|
|
exclude_inaccessible_preferences(structured_preferences, request.user)
|
|
structured_preferences = add_non_editable_in_preference(structured_preferences)
|
|
|
|
return Response({
|
|
'status': 'success',
|
|
'message': 'Notification preferences retrieved successfully.',
|
|
'show_preferences': get_show_notifications_tray(),
|
|
'data': structured_preferences
|
|
}, status=status.HTTP_200_OK)
|
|
|
|
def put(self, request):
|
|
"""
|
|
Handles PUT requests to update notification preferences.
|
|
|
|
This method updates the user's notification preferences based on the
|
|
provided data in the request body. It expects a dictionary with
|
|
notification types and their settings.
|
|
|
|
Returns:
|
|
Response: A DRF Response object indicating success or failure.
|
|
"""
|
|
# Validate incoming data
|
|
serializer = UserNotificationPreferenceUpdateAllSerializer(data=request.data)
|
|
if not serializer.is_valid():
|
|
return Response({
|
|
'status': 'error',
|
|
'message': serializer.errors
|
|
}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
# Get validated data for easier access
|
|
validated_data = serializer.validated_data
|
|
|
|
# Build query set based on notification type
|
|
query_set = NotificationPreference.objects.filter(user_id=request.user.id)
|
|
|
|
if validated_data['notification_type'] == 'grouped_notification':
|
|
# Get core notification types for the app
|
|
grouped_types = filter_notification_types_by_app(validated_data['notification_app'], use_app_defaults=True)
|
|
query_set = query_set.filter(type__in=grouped_types.keys())
|
|
else:
|
|
# Filter by single notification type
|
|
query_set = query_set.filter(type=validated_data['notification_type'])
|
|
|
|
# Prepare update data based on channel type
|
|
updated_data = self._prepare_update_data(validated_data)
|
|
|
|
# Update preferences
|
|
query_set.update(**updated_data)
|
|
|
|
# Log the event
|
|
self._log_preference_update_event(request.user, validated_data)
|
|
|
|
# Prepare and return response
|
|
response_data = self._prepare_response_data(validated_data)
|
|
return Response(response_data, status=status.HTTP_200_OK)
|
|
|
|
def _prepare_update_data(self, validated_data):
|
|
"""
|
|
Prepare the data dictionary for updating notification preferences.
|
|
|
|
Args:
|
|
validated_data (dict): Validated serializer data
|
|
|
|
Returns:
|
|
dict: Dictionary with update data
|
|
"""
|
|
channel = validated_data['notification_channel']
|
|
|
|
if channel == 'email_cadence':
|
|
return {channel: validated_data['email_cadence']}
|
|
else:
|
|
return {channel: validated_data['value']}
|
|
|
|
def _log_preference_update_event(self, user, validated_data):
|
|
"""
|
|
Log the notification preference update event.
|
|
|
|
Args:
|
|
user: The user making the update
|
|
validated_data (dict): Validated serializer data
|
|
"""
|
|
event_data = {
|
|
'notification_app': validated_data['notification_app'],
|
|
'notification_type': validated_data['notification_type'],
|
|
'notification_channel': validated_data['notification_channel'],
|
|
'value': validated_data.get('value'),
|
|
'email_cadence': validated_data.get('email_cadence'),
|
|
}
|
|
notification_preference_update_event(user, [], event_data)
|
|
|
|
def _prepare_response_data(self, validated_data):
|
|
"""
|
|
Prepare the response data dictionary.
|
|
|
|
Args:
|
|
validated_data (dict): Validated serializer data
|
|
|
|
Returns:
|
|
dict: Response data dictionary
|
|
"""
|
|
email_cadence = validated_data.get('email_cadence', None)
|
|
# Determine the updated value
|
|
updated_value = validated_data.get('value', email_cadence if email_cadence else None)
|
|
|
|
# Determine the channel
|
|
channel = validated_data.get('notification_channel')
|
|
if not channel and validated_data.get('email_cadence'):
|
|
channel = 'email_cadence'
|
|
|
|
return {
|
|
'status': 'success',
|
|
'message': 'Notification preferences update completed',
|
|
'show_preferences': get_show_notifications_tray(),
|
|
'data': {
|
|
'updated_value': updated_value,
|
|
'notification_type': validated_data['notification_type'],
|
|
'channel': channel,
|
|
'app': validated_data['notification_app'],
|
|
}
|
|
}
|