셀러리와 장고에서 순환 수입 해결하기

Question

샐러리를 사용하여 장거리 작업을 수행하는 장고 앱이 있습니다. 주로 데이터베이스 테이블의 일부 필드 계산을 연기합니다.

그래서, 나는 tasks.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) 

지금 분명,이 때문에 원형 수입 작동하지 않습니다

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에서을 (models 수입 tasks 및 tasks 수입품 models).

을 views.py에서 불러 와서이 문제를 해결했지만 모델 논리를 모델 클래스 내에 유지하는 것이 좋습니다. 이 작업을 수행하는 더 좋은 방법이 있습니까?

Answer 1

를 사용하여 신호를 호출 할 send_task 방법을 사용할 수 있습니다

.

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에서 동일한 작업을 수행 할 수도 있습니다. 시작하기 전에 모델을 사용하기 바로 전에 모델을 가져올 수 있습니다.

대안 : 당신은 당신의 작업을

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에 의존합니다.

따라서 tasks.py에서 모델에서 가져 오기보다는 apps.get_model()을 사용하여 모델에 액세스 할 수 있습니다.당신은 확실히 그냥 직접하지만 apps.get_model()를 사용할 수

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

: 다음

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) 

를 실행 한 다음,이 고통스러운 이유

나는 문서의 건강한 비트 헬퍼 방법으로이 작업을 수행 할 단지 표현 .