Python: Calculate time between current time and last login. (Automated Communication)

user_9745220 0
Viewed 314

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:

enter image description here

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)
celery django python
2 Answers

You'll have to iterate over your queried users at least once but here are tips that may help:

models.py

class User(...):

    # add a field to determine if the user has registered or not
    # set this to `True` when a User successfully registers:
    is_registered = models.BooleanField(default=False)

class AutomatedCommunicationNotifications(BaseModel):

    # add a related name field for easier coding:
    user = models.ForeignKey(..., related_name = 'notifications')

tasks.py

# load packages outside of your function so this only runs once on startup:
from django.models import F
from apps.users.models import User
from apps.notifications.models import AutomatedCommunicationNotifications

@app.task
def check_registered_users():

    # timestamps:
    two_hours_ago = datetime.now() - timedelta(hours=2)

    # query for unregistered users who have not received a notification:
    users = User.objects.filter(
        is_registered = False,
        last_login__lt = two_hours_ago # last logged in 2 or more hours ago
    ).exclude(
        notifications__type = "the type"
    ).prefetch_related(
        'notifications' # prejoins tables to improve performance
    )

    for user in users:

        # send email
        ...
user_128216750

I would do this with a cron job. You can let it run whenever you want, depends on how fast after your give time frame you want to sent this.

You start with making a folder in your app: /django/yourapp/management/commands

There you make a python file which contains your logic. Make sure to import the right modules from your views.

from django.core.management.base import BaseCommand, CommandError
from yourapp.models import every, module, you, need
from django.utils import timezone
from datetime import datetime, date, timedelta
from django.core.mail import send_mail, EmailMessage

class Command(BaseCommand):
    help = 'YOUR HELP TEXT FOR INTERNAL USE'

    def handle(self, *args, **options):
        # Your logic

I added the crontab to the www-data users crontab like this:

# m h  dom mon dow   command
45 3 * * * /websites/vaccinatieplanner/venv/bin/python /websites/vaccinatieplanner/manage.py reminder

You can use that crontab file to tweak your optimal time between checks. If you remove the 3 and replace it by a * then you will have it check every 45 mins.

user_31645820
Related