9f136a4149
* feat: added notification when response is endorsed or answered * test: added and fixed test cases * fix: fixed lint errors * refactor: changed method name for readibility * feat: added notification when my response is endorsed * test: fixed failed test cases
69 lines
2.9 KiB
Python
69 lines
2.9 KiB
Python
"""
|
|
Contain celery tasks
|
|
"""
|
|
from celery import shared_task
|
|
from django.contrib.auth import get_user_model
|
|
from edx_django_utils.monitoring import set_code_owner_attribute
|
|
from opaque_keys.edx.locator import CourseKey
|
|
from lms.djangoapps.courseware.courses import get_course_with_access
|
|
from openedx.core.djangoapps.django_comment_common.comment_client.thread import Thread
|
|
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS, ENABLE_COURSEWIDE_NOTIFICATIONS
|
|
from lms.djangoapps.discussion.rest_api.discussions_notifications import DiscussionNotificationSender
|
|
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
@shared_task
|
|
@set_code_owner_attribute
|
|
def send_thread_created_notification(thread_id, course_key_str, user_id):
|
|
"""
|
|
Send notification when a new thread is created
|
|
"""
|
|
course_key = CourseKey.from_string(course_key_str)
|
|
if not (ENABLE_NOTIFICATIONS.is_enabled(course_key) and ENABLE_COURSEWIDE_NOTIFICATIONS.is_enabled(course_key)):
|
|
return
|
|
thread = Thread(id=thread_id).retrieve()
|
|
user = User.objects.get(id=user_id)
|
|
course = get_course_with_access(user, 'load', course_key, check_if_enrolled=True)
|
|
notification_sender = DiscussionNotificationSender(thread, course, user)
|
|
notification_sender.send_new_thread_created_notification()
|
|
|
|
|
|
@shared_task
|
|
@set_code_owner_attribute
|
|
def send_response_notifications(thread_id, course_key_str, user_id, parent_id=None):
|
|
"""
|
|
Send notifications to users who are subscribed to the thread.
|
|
"""
|
|
course_key = CourseKey.from_string(course_key_str)
|
|
if not ENABLE_NOTIFICATIONS.is_enabled(course_key):
|
|
return
|
|
thread = Thread(id=thread_id).retrieve()
|
|
user = User.objects.get(id=user_id)
|
|
course = get_course_with_access(user, 'load', course_key, check_if_enrolled=True)
|
|
notification_sender = DiscussionNotificationSender(thread, course, user, parent_id)
|
|
notification_sender.send_new_comment_notification()
|
|
notification_sender.send_new_response_notification()
|
|
notification_sender.send_new_comment_on_response_notification()
|
|
notification_sender.send_response_on_followed_post_notification()
|
|
|
|
|
|
@shared_task
|
|
@set_code_owner_attribute
|
|
def send_response_endorsed_notifications(thread_id, course_key_str, comment_author_id):
|
|
"""
|
|
Send notifications when a response is marked answered/ endorsed
|
|
"""
|
|
course_key = CourseKey.from_string(course_key_str)
|
|
if not ENABLE_NOTIFICATIONS.is_enabled(course_key):
|
|
return
|
|
thread = Thread(id=thread_id).retrieve()
|
|
comment_author = User.objects.get(id=comment_author_id)
|
|
course = get_course_with_access(comment_author, 'load', course_key, check_if_enrolled=True)
|
|
notification_sender = DiscussionNotificationSender(thread, course, comment_author)
|
|
#sends notification to author of thread
|
|
notification_sender.send_response_endorsed_on_thread_notification()
|
|
#sends notification to author of response
|
|
notification_sender.send_response_endorsed_notification()
|