芹菜任务和自定义装饰器
|
我正在使用django和celery(django-celery)进行项目。我们的团队决定将所有数据访问代码包装在ѭ0内(不像ѭ1那样包装到Manager中),而让代码在(app-name)/task.py中只处理用celery组装和执行任务(因此我们不会\'在此层中具有django ORM依赖性)。
在我的
manager.py
中,我有以下内容:
def get_tag(tag_name):
ctype = ContentType.objects.get_for_model(Photo)
try:
tag = Tag.objects.get(name=tag_name)
except ObjectDoesNotExist:
return Tag.objects.none()
return tag
def get_tagged_photos(tag):
ctype = ContentType.objects.get_for_model(Photo)
return TaggedItem.objects.filter(content_type__pk=ctype.pk, tag__pk=tag.pk)
def get_tagged_photos_count(tag):
return get_tagged_photos(tag).count()
在我的task.py中,我喜欢将它们包装成任务(然后可以使用这些任务来完成更复杂的任务),因此我编写了这个装饰器:
import manager #the module within same app containing data access functions
class mfunc_to_task(object):
def __init__(mfunc_type=\'get\'):
self.mfunc_type = mfunc_type
def __call__(self, f):
def wrapper_f(*args, **kwargs):
callback = kwargs.pop(\'callback\', None)
mfunc = getattr(manager, f.__name__)
result = mfunc(*args, **kwargs)
if callback:
if self.mfunc_type == \'get\':
subtask(callback).delay(result)
elif self.mfunc_type == \'get_or_create\':
subtask(callback).delay(result[0])
else:
subtask(callback).delay()
return result
return wrapper_f
然后(仍在task.py
中):
#@task
@mfunc_to_task()
def get_tag():
pass
#@task
@mfunc_to_task()
def get_tagged_photos():
pass
#@task
@mfunc_to_task()
def get_tagged_photos_count():
pass
没有@task
,一切正常。
但是,在使用那个“ 7”型装饰器(按照芹菜文档的指示放到顶部)之后,事情就开始崩溃了。显然,每次调用ѭ9时,都会传递与ѭ11相同的task.get_tag
函数。所以我每次都得到相同的wrapper_f
,现在我唯一要做的就是获得一个标签。
我是装饰工的新手。任何人都可以帮助我了解这里出了什么问题,或者指出完成任务的其他方法?我真的很讨厌为我的每个数据访问功能编写相同的任务包装代码。
没有找到相关结果
已邀请:
2 个回复
掸牛浓疗
让我们向MyCoolTask添加一些日志记录:
并创建一个扩展类(扩展MyCoolTask,但现在带有参数):
并确保将kwargs作为json数据传递:
我得到结果:
凄嘛姥
的基类,而不是使用decorator? 这样,您的所有任务都可以扩展自定义的任务类,您可以在其中使用方法
和
实现个人行为 。 您还可以为所有任务定义通用方法和对象。