这是在为用户刷新访问令牌的上下文中。假设我们有refresh_user_token
函数,它接受CustomUser
对象作为user
。
def refresh_user_token(user):
...
...
return result
我想为每个特定的CustomUser
执行此功能,以便在9天内安排重复。
def refresh_user_token(user):
...
...
next_refresh = datetime.now() + timedelta(days=9)
schedule_refresh(user, scheduled_time=next_refresh)
return result
我在Celery中看到的大多数用例都是关于执行批处理操作,但是对于这种用法,我需要能够使用Celery执行带有参数的函数。
有人建议设置一个cron作业来检查任何需要刷新非常x秒的令牌。
所以在CustomUser
对象上,我们有一个名为DateTimeField
的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
这可能有效,但是当消息代理只能用于安排所需的任务时,我觉得非常费力。
您可以使用Celery Beat使用crontabs安排芹菜任务。只需创建常规Celery任务并使用Beat来说明何时运行。
这是我在项目中的Celery Beat设置示例:
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'}
},
}