在django框架外运行celery任务

ssgvzors  于 2023-05-19  发布在  Go
关注(0)|答案(1)|浏览(261)

我有一个django/celery应用程序,它工作得很好。在正常操作中,任务会按预期创建并插入工作队列。
我试图创建一个单独的Python脚本,将手动运行,这是试图插入一个任务到工人队列...但它不起作用。
测试功能很简单:

@shared_task()
def testtask(num=100, tag=0):
    for i in range(num):
        print_log("in testtask {0} {1}".format(i, tag))
        time.sleep(1)
    return

在python3提示符下,我执行以下操作:

>>> os.environ.setdefault("DJANGO_SETTINGS_MODULE", "raxakprotect.settings.dev")        
>>> import django                                                                 
>>> django.setup()                                                                
>>> from test import testtask 
# test has other django dependencies hence the django.setup()
>>> testtask(10) # This works fine... but
>>> testtask.delay(10) # hangs indefinitely

不知道为什么会这样。我必须以某种方式单独连接到celery 经纪人吗?
观察celery事件屏幕显示未创建任何事件或条目。
谢谢prasanna

wmomyfyw

wmomyfyw1#

希望这能帮助其他犯同样错误的人--我发现我做错了什么。您需要设置celery 应用程序并将其指向其设置,如下所示:

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "raxakprotect.settings.prod")    
import django
django.setup()

from raxakprotect.celery import app
app.config_from_object("django.conf:settings", namespace="CELERY")               
app.autodiscover_tasks()

最后三行在我最初的尝试中丢失了。

相关问题