Django - 使用选项但没有函数参数执行 Celery 任务

Question

我是第一次使用 Celery。查看文档，似乎我已经尝试了一切以使用 apply_async 正确执行任务。我从信号方法调用任务，所以它在signals.py中。

信号.py

from django.db.models.signals import post_save
from django.dispatch import receiver
from datetime import datetime, timedelta

from games.models import Game
from contacts.models import Contact
from msgs.models import SMS
from msgs.tasks import sendSMS_Scheduled


@receiver(post_save, sender=Game)
def createSMS(sender, instance, created, **kwargs):

    if created:

        contacts = Contact.objects.all()
        body = SMS.defaultMessageBuilder(
            location=instance.location,
            time=instance.schedStart
        )
        eta = instance.schedStart - timedelta(hours=7)
        expire = instance.schedStart
        result = sendSMS_Scheduled.apply_async(eta=eta, expires=expire)

tasks.py

from celery import shared_task
from SMSerSite.celery import app
from contacts.models import Contact

from .models import SMS

@shared_task(name='SMSerSite.msgs.tasks.sendSMS', bind=True)
def sendSMS_Scheduled():

    messages = self.request.sms_set.all()

    SMS.sendSMS(messages)

代码运行时出现错误：sendSMS_Scheduled() takes 0 positional arguments but 1 was given。我尝试了各种方法来编写使用 apply_async 调用任务的行，但没有任何效果。我在这里做错了什么？

Answer 1

当您使用 bind=True 关键字参数时，您的任务会获得对其自身的引用，因此请将您的方法签名更改为：

def sendSMS_Scheduled(self):

作为旁注，请记住 self 在这种情况下是实际的 celery 任务。你有一条线 self.request.sms_set.all() 这让我觉得你期待 self 是 django 特定的东西。