Решающий круговой импорт в сельдерей и 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).

Я решил это на данный момент, позвонив my_task.delay() от views.py, но, похоже, имеет смысл сохранить модельную логику в классе модели. Есть ли лучший способ сделать это?

Answer 1

Используйте сигналы.

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=?) 

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

Решение опубликовано Навина очень хорошо, но когда я впервые попробовал это, я обнаружил, что мои @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() является метод на объектах сельдерея приложений.

В этом решении важно выполнить задачи take care of correct, predictable names.

Answer 4

Просто, чтобы бросить еще одно не очень большое решение в этот список, то, что я закончил, опирается на django's now-built-in app registry.

я сделать это с помощью вспомогательного метода со здоровым немного документации просто выразить, почему это больно:

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() непосредственно, хотя ,