芹菜工人和命令行参数答案

【问题标题】：Celery worker and command line args芹菜工人和命令行参数
【发布时间】：2014-01-26 15:15:07
【问题描述】：

我正在重构我的代码以使用 celery worker。

之前我使用 argparse 来传递命令行参数。

例如

if __name__ == "__main__":
     parser = argparse.ArgumentParser(description='Node')
     parser.add_argument('--environment', action="store", default='local', help="env e.g. production of development")
     environment = arg_options.environment

但现在我得到了这个错误。

celery -A tasks worker --loglevel=info --environment local
celery: error: no such option: --environment

如何添加？

如果不需要，我不想使用环境变量。

e.g export environment=development

【问题讨论】：

标签： python celery

【解决方案1】：

Celery worker 不会执行你的__main__。

如果你想添加额外的命令行选项，你可以使用app.user_options，但请注意，它使用optparse 模块，而不是argparse。

有关更多信息，请参阅文档中的此部分：

http://docs.celeryproject.org/en/latest/userguide/extending.html#preload-options

【讨论】：

我了解如何将参数输入到引导步骤 init 方法中，但是如何从任务中访问这些参数？
或者如何获得它们......坦率地说。在这方面损失了几个小时

【解决方案2】：

通过查看源代码，我找到了处理此问题的方法。

在 celery.py 上添加：

from celery import bootsteps
from celery.bin import Option

....
app.user_options['worker'].add(
    Option('--server', dest='api_server', default='127.0.0.1',
           help='API server to use.')
)

app.conf['API_SERVER'] = '127.0.0.1'

class ConfigBootstep(bootsteps.Step):
    def __init__(self, worker, api_server=None, **options):
        app.conf['API_SERVER'] = api_server

app.steps['worker'].add(ConfigBootstep)

然后在保存你的任务的文件中添加：

from celery import current_app
...
@shared_task()
def roboscope():
    API_SERVER = current_app.conf.get('API_SERVER', '127.0.0.1')

我尝试通过在导入模块时解析它来使 API_SERVER 成为模块全局变量，但它不起作用，因为它为时过早。由于我的任务非常密集，因此运行多次并没有什么坏处。

【讨论】：

您可以使用bootsteps.StartStopStep 类作为基础。通过这种方式，您可以更早地获取值（并将其保存在变量或其他东西中），甚至可以在 worker 启动时验证参数的值。
"on celery.py add:" 原谅我 - 但你不应该与 celery 的命名空间发生冲突。选择 'celery.py' 肯定会破坏多模块应用程序。

【解决方案3】：

可以使用worker选项。

例如，我必须在其中一项任务中初始化 mongo 数据库

from celery import bootsteps
from celery.bin import Option
import mongoConfig
.....


app = Celery('scraper')

app.user_options['worker'].add(
    Option('--mongo', dest='is_mongo_required', default=None, help='Mongo Required')
)

class CustomArgs(bootsteps.Step):

    def __init__(self, worker, is_mongo_required=None, **options):
        # store the api authentication
        if is_mongo_required is not None and is_mongo_required[0] == "true":
            print("Mongo initialization required")
            mongoConfig.init(required=True)


app.steps['worker'].add(CustomArgs)


app.config_from_object('celeryConfig')

要执行的命令

celery worker -A scraper --concurrency=4 --loglevel=info -Q abc,bcd --mongo=true

【讨论】：