python - 如何使用 MySQL 中的日期时间在 Celery 中执行任务？

Question

新闻以发布的日期时间存储在数据库MySQL中。

任何时候用户都可以在延迟日期发布的表中添加新的。

如何使用Celery 列出数据库表并检查是否到了发布数据的时间？

负责发布数据的另一个进程(脚本 Python)。所以 Celery 应该根据 datetime 为 MySQL 表中的每一行调用这个脚本。

如何使用 Celery 做到这一点？

我认为另一种方法是使用发布的 date 和 id 创建队列，然后直接将来自用户表单的数据添加到该队列中。因此，应该是进程(Celery)，它观察队列以按日期进一步执行任务。

Celery 提供了 Periodic tasks 和 Scheduled tasks 的文档，但是没有解释和示例如何使用它。

抱歉，如果问题很简单，我是 Python Celery 的新手。

Answer 1

您可以通过在apply_async() 上使用参数eta 来执行任务(引用http://docs.celeryproject.org/en/latest/userguide/calling.html#eta-and-countdown)

eta 参数必须是日期时间格式，因此它将以精确的毫秒精度运行。

我建议你使用 ORM，这样你数据库中的日期时间数据类型将由 ORM 自动转换 :D

假设我们有一个文章模型:

class Article(models.Model):
    title = models.CharField(max_length=100)
    published = models.BooleanField(default=False)
    created_date = models.DateTimeField()

然后是我们的任务模块:

@app.task(bind=True)
def set_published(*args, **kwargs):
    article = Article.objects.get(id=args[1])
    article.published = True
    article.save()

    print("article %d has been published" % args[1])

保存新文章并调用 set_published ETA+1minute

article = Article(title='This is a new article', published=False, created_date=datetime.utcnow())
article.save()

delta = timedelta(minutes=1)
set_published.apply_async(args=(article.id,), eta=article.created_date + delta)