Schedule Task for User 9 Days After Last Refresh Django/Celery

Question

This is in the context of refreshing an access token for a user. Say we have the function refresh_user_token that takes in a CustomUser object as user.

def refresh_user_token(user):
    ...
    ...
    return result

I'd like for each execution of this function for a specific CustomUser to schedule a repeat in 9 days.

def refresh_user_token(user):
    ...
    ...
    next_refresh = datetime.now() + timedelta(days=9)
    schedule_refresh(user, scheduled_time=next_refresh)
    return result

Most use cases I see for Celery are in regards to performing batch operations, but for this use I need to be able to execute the function with arguments which doesn't seem to be feasible with Celery.

---

Someone did recommend setting up a cron job to check for any tokens that need refreshed very x seconds.

So on the CustomUser object, we have a DateTimeField called last_token_refresh.

@Celery.task
def refresh_auth_tokens():
    users = CustomUser.objects.all()
    for user in users:
        last_refresh_delta = datetime.now(timezone.utc) - user.last_token_refresh
        if last_refresh_delta.days >= 9:
            refresh_user_token(user)
            return True
        else:
            return False

This could work, but I feel its very taxing when a message broker could be used to only schedule the tasks needed.

Answer 1

You can use Celery Beat to schedule celery tasks using crontabs. Just create your regular Celery tasks and use Beat to say when to run.

This is a sample of a Celery Beat setup I have on a project:

CELERY_BEAT_SCHEDULE = {
    'populate_controller': {
        'task': 'common.tasks.populate_controller',
        # Will be executed Mondays, at 08:30 
        'schedule': crontab(day_of_week=2, 
                            hour=8, 
                            minute=30),
    'options': {'queue': 'populate_controller'}
    },
 }