generic_celery_task 0.2

问题

解决方案

只使用一个通用到可以运行任意python的芹菜任务 具有任意参数的函数，并将此任务包装为自定义 装饰工。

作为奖励，当芹菜/ 代理未运行。

作业的调用API与芹菜的相同：.s（），.delay（）和 .apply_async（）

示例

• 在与celleryconfig.py位于同一目录的celleryapp.py中， 创建一个芹菜应用程序，然后导入自定义装饰程序：

importceleryimportosimportsyssys.path.insert(0,os.path.join(os.path.dirname(__file__),'.'))#os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'settings')app=celery.Celery()app.config_from_object('celeryconfig')fromgeneric_celery_task.decoratorsimporttask

• 和经纪人一起做芹菜你不再需要重新启动 之后是芹菜
• 在另一个文件中，importable from celleryapp.py，创建您的任务：

fromceleryappimporttask@taskdefjob(x,y):return'x + y = %d'%(x+y)

• 现在使用它：

importcelery# direct function callassertjob(1,2)=='x + y = 3'# using the delay() methodres=job.delay(1,2)# if the Celery daemon and its backend broker are running, 'res' is an instance of AsyncResultifisinstance(res,celery.result.AsyncResult):assertres.get()=='x + y = 3'else:# if either one is not running, a warning is issued and the function is executed synchronously# with 'res' being the actual function result# if you want to silence the warnings, use '@task(quiet=True)'assertres=='x + y = 3'# using the apply_async() methodres=job.apply_async(args=[1,2])# and process the result as above, if you need to

安装

提供setup.py你知道怎么处理

测试

测试需要nose、redis、redis py并假设端口6389是 免费的。

使用“python setup.py test”或“notests-v”运行测试

这个包是用python-2.7.6、python-3.3.4、nose-1.3.0和 芹菜-3.1.10，redis-2.8.7和redis-py-2.9.1。

注意事项

• 保存自定义任务的模块将被重新加载。如果它 包含使用“super”的类及其实例，您可能会遇到 所描述的问题 here。 应用一个建议的修复
• 芹菜守护进程及其代理的状态只检查一次， 当第一个.delay（）或.apply_async（）方法在 自定义任务。

学分

• 作者：stefan talpalaru stefantalpalaru@yahoo.com
• 主页：https://github.com/stefantalpalaru/generic_celery_task