解决 celery 和 django 中的循环导入问题

Question

我有一个 Django 应用程序，它使用 Celery 卸载一些任务。主要是延迟了数据库表中某些字段的计算。

所以，我有一个tasks.py：

from models import MyModel
from celery import shared_task

@shared_task
def my_task(id):
    qs = MyModel.objects.filter(some_field=id)
    for record in qs:
        my_value = #do some computations
        record.my_field = my_value
        record.save()

在models.py中

 from django.db import models
 from tasks import my_task

 class MyModel(models.Model):
      field1 = models.IntegerField()
      #more fields
      my_field = models.FloatField(null=True)

      @staticmethod
      def load_from_file(file):
          #parse file, set fields from file
          my_task.delay(id)

现在显然，由于循环导入（models 导入 tasks 和 tasks 导入 models），这将不起作用。

我暂时通过从views.py 调用my_task.delay() 解决了这个问题，但是将模型逻辑保留在模型类中似乎是有意义的。有更好的方法吗？

Answer 1

joshua 贴的解决方案很好，但是我第一次尝试的时候发现我的@receiver装饰器没有效果。那是因为 tasks 模块没有在任何地方导入，这是预期的，因为我使用了 task auto-discovery。

但是，还有另一种方法可以将 tasks.py 与 modules.py 分离。也就是说，任务可以按名称发送，而不必在发送它们的过程中评估（导入）：

from django.db import models
#from tasks import my_task
import celery

class MyModel(models.Model):
    field1 = models.IntegerField()
    #more fields
    my_field = models.FloatField(null=True)

    @staticmethod
    def load_from_file(file):
        #parse file, set fields from file
        #my_task.delay(id)
        celery.current_app.send_task('myapp.tasks.my_task', (id,))

send_task() 是 Celery 应用对象上的一个方法。

在此解决方案中，take care of correct, predictable names 对您的任务很重要。

Answer 2

在您的模型中，您可以在使用它之前导入它，而不是在文件开头导入my_task。它将解决循环导入问题。

from django.db import models

class MyModel(models.Model):
      field1 = models.IntegerField()
      #more fields
      my_field = models.FloatField(null=True)

      @staticmethod
      def load_from_file(file):
          #parse file, set fields from file
          from tasks import my_task   # import here instead of top
          my_task.delay(id)

或者，您也可以在 tasks.py 中执行相同的操作。您可以在使用之前导入模型，而不是开始。

替代方案：

您可以使用send_task 方法来调用您的任务

from celery import current_app
from django.db import models

class MyModel(models.Model):
      field1 = models.IntegerField()
      #more fields
      my_field = models.FloatField(null=True)

      @staticmethod
      def load_from_file(file):
          #parse file, set fields from file
          current_app.send_task('myapp.tasks.my_task', (id,))

Answer 3

只是为了在这个列表中再添加一个不太好的解决方案，我最终做的是依赖django's now-built-in app registry。

所以在tasks.py 中，您可以使用apps.get_model() 来访问模型，而不是从模型中导入。

我用一个带有健康文档的辅助方法来做这件事，只是为了表达为什么这是痛苦的：

from django.apps import apps

def _model(model_name):
    """Generically retrieve a model object.

    This is a hack around Django/Celery's inherent circular import
    issues with tasks.py/models.py. In order to keep clean abstractions, we use
    this to avoid importing from models, introducing a circular import.

    No solutions for this are good so far (unnecessary signals, inline imports,
    serializing the whole object, tasks forced to be in model, this), so we
    use this because at least the annoyance is constrained to tasks.
    """
    return apps.get_model('my_app', model_name)

然后：

@shared_task
def some_task(post_id):
    post = _model('Post').objects.get(pk=post_id)

你当然可以直接使用apps.get_model()。

Answer 4

使用信号。

tasks.py

from models import MyModel, my_signal
from celery import shared_task
from django.dispatch import receiver

@shared_task
def my_task(id):
    qs = MyModel.objects.filter(some_field=id)
    for record in qs:
        my_value = #do some computations
        record.my_field = my_value
        record.save()

@receiver(my_signal)
def my_receiver(sender, **kwargs):
    my_task.delay(kwargs['id'])

models.py

 from django.db import models
 from tasks import my_task
 from django.dispatch import Signal

 my_signal = Signal(providing_args=['id'])

 class MyModel(models.Model):
      field1 = models.IntegerField()
      #more fields
      my_field = models.FloatField(null=True)

      @staticmethod
      def load_from_file(file):
          #parse file, set fields from file
          my_signal.send(sender=?, id=?)