ammar/zod-backend
1
0
Fork 0
mirror of https://github.com/HamzaSha1/zod-backend.git synced 2025-07-10 23:27:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

changed notification method, added celery task to notify expiring task

Browse Source
This commit is contained in:
abutalib-kiwi
2023-08-24 12:10:18 +05:30
parent 5b9f7b1828
commit f541608656
12 changed files with 153 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
notifications/utils.py
View File

@ -11,7 +11,8 @@ from django.contrib.auth import get_user_model
from account.models import UserNotification
from account.utils import get_user_full_name
from guardian.models import Guardian
from base.constants import GUARDIAN, JUNIOR
from guardian.models import Guardian, JuniorTask
from junior.models import Junior
from notifications.constants import NOTIFICATION_DICT
from notifications.models import Notification
@ -42,44 +43,64 @@ def remove_fcm_token(user_id: int, access_token: str, registration_id) -> None:
print(e)
def get_basic_detail(from_user_id, to_user_id):
def get_from_user_details(from_user_id, from_user_type):
"""
used to get the basic details
used to get from user details
"""
from_user = User.objects.get(id=from_user_id) if from_user_id else None
to_user = User.objects.get(id=to_user_id)
return from_user, to_user
from_user = None
from_user_name = None
from_user_image = None
if from_user_id:
if from_user_type == GUARDIAN:
guardian = Guardian.objects.filter(user_id=from_user_id).select_related('user').first()
from_user = guardian.user
from_user_name = get_user_full_name(from_user)
from_user_image = guardian.image
elif from_user_type == JUNIOR:
junior = Junior.objects.filter(auth_id=from_user_id).select_related('auth').first()
from_user = junior.auth
from_user_name = get_user_full_name(from_user)
from_user_image = junior.image
return from_user_name, from_user_image, from_user
def get_notification_data(notification_type, from_user, to_user, extra_data):
def get_notification_data(notification_type, from_user_id, from_user_type, to_user_id, extra_data):
"""
get notification and push data
:param from_user_type: GUARDIAN or JUNIOR
:param notification_type: notification_type
:param from_user: from_user obj
:param to_user: to_user obj
:param from_user_id: from_user obj
:param to_user_id: to_user obj
:param extra_data: any extra data provided
:return: notification and push data
"""
push_data = NOTIFICATION_DICT[notification_type].copy()
notification_data = push_data.copy()
notification_data['to_user'] = get_user_full_name(to_user)
if from_user:
from_user_name = get_user_full_name(from_user)
push_data['body'] = push_data['body'].format(from_user=from_user_name)
notification_data['body'] = notification_data['body'].format(from_user=from_user_name)
notification_data['from_user'] = from_user_name
task_name = None
if 'task_id' in extra_data:
task = JuniorTask.objects.filter(id=extra_data.get('task_id')).first()
task_name = task.task_name
extra_data['task_image'] = task.image if task.image else task.default_image
from_user_name, from_user_image, from_user = get_from_user_details(from_user_id, from_user_type)
push_data['body'] = push_data['body'].format(from_user=from_user_name, task_name=task_name)
notification_data['body'] = notification_data['body'].format(from_user=from_user_name, task_name=task_name)
notification_data['from_user'] = from_user_name
notification_data['from_user_image'] = from_user_image
notification_data.update(extra_data)
return notification_data, push_data
to_user = User.objects.get(id=to_user_id)
return notification_data, push_data, from_user, to_user
def send_notification(notification_type, from_user, to_user, extra_data):
@shared_task()
def send_notification(notification_type, from_user_id, from_user_type, to_user_id, extra_data):
"""
used to send the push for the given notification type
"""
# (from_user, to_user) = get_basic_detail(from_user_id, to_user_id)
notification_data, push_data = get_notification_data(notification_type, from_user, to_user, extra_data)
notification_data, push_data, from_user, to_user = get_notification_data(notification_type, from_user_id,
from_user_type, to_user_id, extra_data)
user_notification_type = UserNotification.objects.filter(user=to_user).first()
notification_data.update({'badge': Notification.objects.filter(notification_to=to_user, is_read=False).count()})
Notification.objects.create(notification_type=notification_type, notification_from=from_user,
@ -104,14 +125,10 @@ def send_notification_to_guardian(notification_type, from_user_id, to_user_id, e
:param extra_data:
:return:
"""
from_user = None
if from_user_id:
from_user = Junior.objects.filter(auth_id=from_user_id).select_related('auth').first()
from_user = Junior.objects.filter(auth_id=from_user_id).first()
extra_data['from_user_image'] = from_user.image
from_user = from_user.auth
to_user = Guardian.objects.filter(user_id=to_user_id).select_related('user').first()
extra_data['to_user_image'] = to_user.image
send_notification(notification_type, from_user, to_user.user, extra_data)
send_notification(notification_type, from_user_id, to_user_id, extra_data)
@shared_task()
@ -123,13 +140,9 @@ def send_notification_to_junior(notification_type, from_user_id, to_user_id, ext
:param extra_data:
:return:
"""
from_user = None
if from_user_id:
from_user = Guardian.objects.filter(user_id=from_user_id).select_related('user').first()
from_user = Guardian.objects.filter(user_id=from_user_id).first()
extra_data['from_user_image'] = from_user.image
from_user = from_user.user
to_user = Junior.objects.filter(auth_id=to_user_id).select_related('auth').first()
extra_data['to_user_image'] = to_user.image
send_notification(notification_type, from_user, to_user.auth, extra_data)
send_notification(notification_type, from_user_id, to_user_id, extra_data)