当任何celery任务入队时,我想添加工作人员能够使用的上下文元数据。
以下代码示例有效,但我希望有一个合适的芹菜式解决方案。
from celery.signals import before_task_publish, task_prerun
@before_task_publish.connect
def receiver_before_task_publish(sender=None, headers=None, body=None, **kwargs):
task_kwags = body[1]
metadata = {"foo": "bar"}
task_kwags['__metadata__'] = metadata
@task_prerun.connect
def receiver_task_pre_run(task_id, task, *args, **kwargs):
metadata = kwargs['kwargs'].pop('__metadata__', {})
# metadata == {"foo": "bar"}
当一个任务在工人中开始时,before_task_publish
的header
的内容位于**kwargs
的push_request
中。
def push_request(self, *args, **kwargs):
self.request_stack.push(Context(*args, **kwargs))
在Context
的构造函数中完成了一些不错的事情。 self.__dict__.update()
意味着我们可以像Context(metadata={'foo': 'bar'}).metadata
一样访问这些值
class Context(object)
# ...
def __init__(self, *args, **kwargs):
self.update(*args, **kwargs)
def update(self, *args, **kwargs):
return self.__dict__.update(*args, **kwargs)
可以从Task
的request
财产访问任务上下文。
class Task(object):
# ...
def _get_request(self):
"""Get current request object."""
req = self.request_stack.top
if req is None:
# task was not called, but some may still expect a request
# to be there, perhaps that should be deprecated.
if self._default_request is None:
self._default_request = Context()
return self._default_request
return req
request = property(_get_request)
这意味着最终的解决方案就是:
from celery.signals import before_task_publish, task_prerun
@before_task_publish.connect
def receiver_before_task_publish(sender=None, headers=None, body=None, **kwargs):
metadata = {"foo": "bar"}
headers['__metadata__'] = metadata
@task_prerun.connect
def receiver_task_pre_run(task_id, task, *args, **kwargs):
metadata = getattr(task.request, '__metadata__', {})
# metadata == {"foo": "bar"}
注意:task.request.__metadata__
也可以工作,但如果任务在信号集成之前排队,它就会失败。这样更安全。