如何从 models.py 导入函数来运行 cron 作业？

Question

我正在尝试使用 celery 运行 cron 作业。问题是我需要在 models.py 中运行一个函数，这听起来很奇怪，但让我在代码中演示一下。

class Post(models.Model):
rank_score = models.FloatField(default= 1)
image = models.ImageField(upload_to='images',blank=True, null=True)


def get_score(self):
    upvote_count = self.vote_set.filter(is_up=True).count()
    devote_count = self.vote_set.filter(is_up=False).count()
    return hot(upvote_count, devote_count, self.pub_date.replace(tzinfo=None)

)

这里我需要每秒在后台运行一次 get_score。所以对于 celery.py 我要导入这个函数。

from  main.models import get_score

logger = get_task_logger(__name__)

@periodic_task(
    run_every=(crontab(minute='*/15')),
    name="task_get_score",
    ignore_result=True
)

def task_get_score():
    get_score()
    logger.info("running cron job")

然后它说 from main.models import get_score ImportError: 无法导入名称 get_score

或者我是否需要在我的 utils 文件夹中的 rank.py 中运行公式函数

def hot(ups, downs, date):

    s = score(ups, downs)
    order = log(max(abs(s), 1), 10)
    sign = 1 if s > 0 else -1 if s < 0 else 0
    seconds = epoch_seconds(date) - 1134028003
    return round(sign * order + seconds / 45000, 7)

这是我应该运行 cron 作业的那个吗？

Answer 1

首先在脚本中设置 djnago 项目，然后使用 get_score 方法

import os
import sys
import django
sys.path.append('/var/www/projects/project_name/')

os.environ['DJANGO_SETTINGS_MODULE'] = 'project_name.settings'
from django.conf import settings

from  main.models import get_score

logger = get_task_logger(__name__)

@periodic_task(
    run_every=(crontab(minute='*/15')),
    name="task_get_score",
    ignore_result=True
)

def task_get_score():
    get_score()
    logger.info("running cron job")