如何解决Django Celery,不执行计划任务
我正在开发一个 Django 应用程序来跟踪加密资产,我想要做的是每 5 天(X 次)拍摄一次用户总价值的快照。
我现在做了什么:
core/celery.py
from __future__ import absolute_import,unicode_literals
import os
from celery import Celery
from celery.schedules import crontab
# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE','core.settings')
app = Celery('core')
# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
# should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings',namespace='CELERY')
app.conf.beat_schedule = {
'print': {
'task': 'app.tasks.test','schedule': 15
},}
app.conf.timezone = 'UTC'
# Load task modules from all registered Django app configs.
app.autodiscover_tasks()
@app.task(bind=True)
def debug_task(self):
print('Request: {0!r}'.format(self.request))
core/settings.py:
# -*- encoding: utf-8 -*-
"""
copyright (c) 2019 - present AppSeed.us
"""
import os
from decouple import config
from unipath import Path
import dj_database_url
import django
# Build paths inside the project like this: os.path.join(BASE_DIR,...)
BASE_DIR = Path(__file__).parent
CORE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# Security WARNING: keep the secret key used in production secret!
SECRET_KEY = config('SECRET_KEY',default='S#perS3crEt_1122')
# Security WARNING: don't run with debug turned on in production!
DEBUG = config('DEBUG',default=True,cast=bool)
# load production server from .env
ALLOWED_HOSTS = ['localhost','127.0.0.1','e6b5f240dedb.ngrok.io',config('SERVER',default='127.0.0.1')]
# Application deFinition
INSTALLED_APPS = [
'django.contrib.admin','django.contrib.auth','django.contrib.contenttypes','django.contrib.sessions','django.contrib.messages','django.contrib.staticfiles','django.contrib.sites','django_celery_beat','chartjs',"allauth","allauth.account","allauth.socialaccount","allauth.socialaccount.providers.google",'app'
]
SITE_ID = 2
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware','whitenoise.middleware.WhiteNoiseMiddleware','django.contrib.sessions.middleware.SessionMiddleware','django.middleware.common.CommonMiddleware','django.middleware.csrf.CsrfViewMiddleware','django.contrib.auth.middleware.AuthenticationMiddleware','django.contrib.messages.middleware.MessageMiddleware','django.middleware.clickjacking.XFrameOptionsMiddleware',]
ROOT_URLconf = 'core.urls'
LOGIN_REDIRECT_URL = "/" # Route defined in app/urls.py
logoUT_REDIRECT_URL = "/login" # Route defined in app/urls.py
TEMPLATE_DIR = os.path.join(
CORE_DIR,"core/templates") # ROOT dir for templates
SOCIALACCOUNT_PROVIDERS = {
'google': {
'ScopE': [
'profile','email',],'AUTH_ParaMS': {
'access_type': 'online',}
}
}
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates','Dirs': [TEMPLATE_DIR],'APP_Dirs': True,'OPTIONS': {
'context_processors': [
'django.template.context_processors.debug','django.template.context_processors.request','django.contrib.auth.context_processors.auth','django.contrib.messages.context_processors.messages',},]
Wsgi_APPLICATION = 'core.wsgi.application'
# Celery Configuration Options
CELERY_TIMEZONE = "Europe/Sofia"
# Database
# https://docs.djangoproject.com/en/3.0/ref/settings/#databases
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.sqlite3','NAME': 'db.sqlite3',}
}
# Password validation
# https://docs.djangoproject.com/en/3.0/ref/settings/#auth-password-validators
AUTH_PASSWORD_VALIDATORS = [
{
'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator',{
'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator',{
'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator',{
'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator',]
# Cron jobs
# Internationalization
# https://docs.djangoproject.com/en/3.0/topics/i18n/
LANGUAGE_CODE = 'en-us'
TIME_ZONE = 'UTC'
USE_I18N = True
USE_L10N = True
USE_TZ = True
#############################################################
# SRC: https://devcenter.heroku.com/articles/django-assets
# Static files (CSS,JavaScript,Images)
# https://docs.djangoproject.com/en/1.9/howto/static-files/
STATIC_ROOT = os.path.join(CORE_DIR,'staticfiles')
STATIC_URL = '/static/'
# Extra places for collectstatic to find static files.
STATICFILES_Dirs = (
os.path.join(CORE_DIR,'core/static'),)
AUTHENTICATION_BACKENDS = (
"django.contrib.auth.backends.ModelBackend","allauth.account.auth_backends.AuthenticationBackend",)
#############################################################
#############################################################
app/tasks.py:
from __future__ import absolute_import,unicode_literals
from celery import shared_task
@shared_task
def test():
print("Message for me and for you")
之后,当我执行
celery -A core worker -l info
我收到错误“芹菜:找不到命令”
注意:我执行了 python manage.py migrate,并且我能够在管理页面中看到新表。 当我运行项目时,没有显示错误,但任务没有执行。
谢谢!
解决方法
您的服务器上似乎还没有安装 Django Celery,请运行
pip install django-celery
然后重新启动一切
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。