Я запускаю Celery / Django на своем удаленном сервере Ubuntu через супервизор. Моя система может успешно получать и выполнять мои 2 задачи от моего 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',
...
]
EDIT:
Я добавил tasks.py к своему post модулю и изменил свой settings.py соответственно выше:
сообщение / 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