I'm trying to make a celery task that would send a basic reminder to our users. So in our automated communication project, we have these tasks:
As you can see there are few actions that are different. So for now I have created a logic that fetches all the users from the DB and then continues by checking the time difference. But for now, I only have set-up for 2 hours or more. How should I use it correctly? I do not want to re-write each if statement because it's bad practice. How should I make it clear and reduce the system load?
@app.task
def check_registered_users():
from apps.users.models import User
from apps.notifications.models import AutomatedCommunicationNotifications
day_start = datetime.utcnow().date()
day_end = day_start + timedelta(days=1)
users = User.objects.filter(is_active=True, date_joined__range=(day_start, day_end))
users_that_received_notification = AutomatedCommunicationNotifications.objects.all().values('user__id')
excluded_users = users.exclude(id__in=users_that_received_notification)
for user in excluded_users:
if user.last_login < user.last_login + timedelta(hours=2):
# Sign-up uncompleted Push notification 2 hours after last login
template = SiteConfiguration.get_solo().automated_comms_signup_uncompleted
send_plain_email_task(
email=user.email,
subject=template.subject,
html_message=template.content,
from_email=f'{settings.EMAIL_FROM_PREFIX} <{settings.DEFAULT_FROM_EMAIL}>',
)
P.S AutomatedCommunicationNotifications table is for us to track which user has already received a notification.
class AutomatedCommunicationNotifications(BaseModel):
""" Model for automated comms notifications """
user = models.ForeignKey(User, on_delete=models.CASCADE)
type = models.CharField(
max_length=255,
choices=NotificationTypes.get_choices(),
default=NotificationTypes.EMAIL_NOTIFICATION
)
def __str__(self):
return str(self.user.phone)
from Python: Calculate time between current time and last login. (Automated Communication)

No comments:
Post a Comment