无法在我的芹菜任务中导入模型

Question

我正在通过主管在我的远程 Ubuntu 服务器上运行 Celery/Django。我的系统可以从我的tasks.py 成功接收并执行我的两个任务。

@periodic_task(run_every=timedelta(minutes=1))
def test_job():
    from polls.models import Question
    from post.models import Post #when I add this line, it fires the error
    for i in Question.objects.all():
        if i.question_text == "test":
            i.question_text = "not_test"
            i.save()
    return HttpResponseRedirect('/')


@periodic_task(name='run_scheduled_jobs', run_every=timedelta(seconds=30))
def run_scheduled_jobs():
    return True

但是，当我使用：

from post.models import Post

在我的任务中，任务失败：

我已经尝试从我的所有其他模块中导入模型，它工作正常；例如from Comment.models import Comment 和 from poll.models import Question 工作正常。出于某种原因，它不允许我导入from post.models import Post。

这是post.models 文件：

from django.db import models
from django.contrib.auth.models import AbstractUser, User
from django.conf import settings
#from django.contrib.humanize import naturaltime
from django.apps import apps
from draft1.choices import CATEGORY_CHOICES
from django.utils import timezone
import requests
import tempfile
from django.core import files
from django.core.files import File as FileWrapper
from datetime import datetime, timedelta
from math import log10
from urllib import parse
from urllib.parse import urlparse
import boto3
# import uuid
from functions.helper_functions import random_string
from math import log, exp
from draft1.choices import DURATION_CHOICES


class Post(models.Model):
    hash = models.CharField(max_length=18, default=random_string, null=True, blank=True)
    user = models.ForeignKey(User, blank=True, null=True)
    has_upvoted = models.ManyToManyField(User, related_name="has_upvoted", blank=True)
    has_downvoted = models.ManyToManyField(User, related_name="has_downvoted", blank=True)
    title = models.TextField(max_length=95)
    second_title = models.TextField(max_length=95, blank=True, null=True)
    dots1 = models.CharField(max_length=90, blank=True)
    dots2 = models.CharField(max_length=90, blank=True)
    dots3 = models.CharField(max_length=90, blank=True)
    last_modified = models.DateTimeField(auto_now=True)
    date = models.DateTimeField(auto_now_add=True)
    content = models.TextField(null=True, blank=True)
    image = models.FileField(null=True, blank=True)
    imageURL = models.URLField(null=True, blank=True)
    video = models.BooleanField(default=False)
    thumbnail = models.ImageField(null=True, blank=True)
    updated = models.DateTimeField(auto_now=True)
    entered_category = models.CharField(max_length=80, default='news')
    ad = models.BooleanField(default=False)
    total_comments = models.IntegerField(default=0)
    post_score = models.ForeignKey('post.PostScore', related_name='post_score', blank=True, null=True)
    hot = models.IntegerField(default=0)

    def handle_uploaded_file(f, filename):
        with open('/tmp/%s' % filename, 'wb+') as destination:
            for chunk in f.chunks():
                destination.write(chunk)

    @property
    def video_source(self):
        if self.imageURL:
            t = urlparse(self.imageURL).netloc
            domain = '.'.join(t.split('.')[1:])
            if domain == "youtube.com":
                return "youtube"
            else:
                return "standard"

    def __str__(self):
        return self.title


class AdvertisePost(Post):
    url = models.CharField(max_length=200, blank=True, null=True)
    position = models.IntegerField(default=DURATION_CHOICES[2][0], choices=DURATION_CHOICES)
    duration = models.IntegerField(default=48)
    total_price = models.IntegerField(default=20)
    activated = models.BooleanField(default=False)
    counter = models.IntegerField(default=0, null=True, blank=True)

    def __str__(self):
        return self.title


class PostScore(models.Model):
    user = models.ForeignKey(User, blank=True, null=True)
    post = models.ForeignKey(Post, related_name='score')
    upvotes = models.IntegerField(default=0)
    downvotes = models.IntegerField(default=0)

    def hot(self):
        s = self.upvotes
        baseScore = log(max(s, 1))
        now = datetime.now()

        timeDiff = (now - self.post.date).days

        if (timeDiff > 1):
            x = timeDiff - 1
            baseScore = baseScore * exp(-8 * x * x)

        return baseScore


class Room(models.Model):
    name = models.CharField(max_length=68)
    frequency = models.IntegerField(default=1)
    comments = models.IntegerField(default=0)
    created = models.DateTimeField(auto_now_add=True)

知道为什么它不允许我导入post.models吗？

PS：模块post.models 肯定存在，当我从其他文件（例如views.py 等）导入它时它工作正常。

settings.py

CELERY_BEAT_SCHEDULER = "django_celery_beat.schedulers.DatabaseScheduler"
CELERYBEAT_SCHEDULE = {
    'run_scheduled_jobs': {
        'task': 'run_scheduled_jobs', 
        'schedule': timedelta(seconds=45),
    },
    'test_job': {
            'task': 'tasks.test_job',
            'schedule': timedelta(seconds=45),
    },
    'post_jobs': {
        'task': 'post.tasks.post_jobs',  
        'schedule': timedelta(minutes=1),
    },
    'test_post': {
        'task': 'post.tasks.test_post',
        'schedule': timedelta(seconds=45),
    }
}

INSTALLED_APPS = [
    ...
    'post',
    ...
]

编辑：

我已将tasks.py 添加到我的post 模块并在上面相应地更改了我的settings.py：

post/tasks.py

@periodic_task(run_every=timedelta(minutes=1))
def test_post():
    from polls.models import Question
    from .models import Post
    for i in Post.objects.all():
        if i.entered_category == "test":
            i.entered_category = "not_test"
            i.save()
    return HttpResponseRedirect('/')


@periodic_task(name='post_jobs', run_every=timedelta(seconds=30)) # task name found! celery will do its job
def post_jobs():
    # do whatever stuff you do
    return True

Answer 1

可能您的 tasks.py 文件不在 django 管理的文件夹下。在 django 项目外部使用 django 相关模块（作为模型）时，我遇到了此类问题。

您的选择很少：

在您的任何 django 模块（应用程序）中实现您的 tasks.py，即：

  ├─┬ post
  │ ├── __init__.py
  │ ├── tasks.py
  │ ├── views.py
  │ └── urls.py
  └── otherapp

在 django 项目中为您的 celery 相关逻辑创建一个模块子文件夹：

  ├─┬ celery
  │ ├── __init__.py
  │ └── tasks.py
  └── post

您还可以将 django 模型导入到您的 tasks.py 中，仍然是 django 项目的外部。看看this的问题。

注意__init__.py 文件，python 需要将文件夹内容作为模块加载。

希望对你有帮助！

更新

我明白了。 最好在文件开头设置导入。您正在将Post 加载到在芹菜工人的共享内存中执行的代码中。这与 django 应用程序管理的内存不同。

尝试在文件的开头进行导入：

from polls.models import Question
from post.models import Post # added post here

@periodic_task(run_every=timedelta(minutes=1))
def test_post():        
    for i in Post.objects.all():
        if i.entered_category == "test":
            i.entered_category = "not_test"
            i.save()
    return HttpResponseRedirect('/')


@periodic_task(name='post_jobs', run_every=timedelta(seconds=30)) # task name found! celery will do its job
def post_jobs():
    # do whatever stuff you do
    return True

如果是这样的话，它会起作用。