我正在做一个使用django和芹菜(django芹菜)的项目。我们的团队决定在(app-name)/manager.py
中包装所有数据访问代码(而不是像django
那样包装到管理器中),并让代码进入(app name)/task.py,只处理用芹菜组装和执行的任务(因此我们在这一层没有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
一切都很好。
但是,在应用了@task
decorator(按照芹菜文档的说明放在顶部)之后,事情就开始崩溃了。显然,每次调用mfunc_to_task.__call__
时,相同的task.get_tag
函数都作为f
传递。所以我每次都得到相同的标签,现在我唯一能做的就是得到一个标签。
我是新来的装修师。有谁能帮助我理解这里出了什么问题,或者指出其他方法来完成任务?我真的不喜欢为我的每个数据访问函数编写相同的任务包装代码。
为什么不创建扩展
celery.Task
的基类而不是使用decorator?这样,所有任务都可以扩展自定义任务类,在该类中,您可以使用方法
__call__
和after_return
实现您的个人行为 . 还可以为所有任务定义通用方法和对象。不太清楚为什么传递论点行不通?
如果使用此示例:
让我们向mycoltask添加一些日志:
并创建一个扩展类(扩展mycoltask,但现在使用参数):
并确保将kwargs作为json数据传递:
我得到的结果是:
相关问题 更多 >
编程相关推荐