Django Celery 结果将任务 ID 设置为人类可读的内容？

Posted 2023-02-24

【中文标题】Django Celery 结果将任务 ID 设置为人类可读的内容？

【英文标题】：Django Celery results set task id to something human readable?

【发布时间】：2018-03-10 06:51:02

【问题描述】：

很多天后，我有了一个工作 celery 和 celery beat 任务列表，结果使用 django_celery_results 存储。但是，当我查看表记录时，其中没有任何有用的信息。

是否可以将任务 ID 设置为人类可读的内容？

一个例子是使用演示任务，它返回一个不可读但不可读的任务 ID

tasks.py

@app.task
def test(a,b):
    return a + b

app.settings 中的调度程序

CELERYBEAT_SCHEDULE = 
    'test_task': 
        'task': 'home.tasks.test',
        'schedule': crontab(minute='*/1'),
    ,

【参考方案1】：

简单的答案是否定的。task_id 属性是自动生成的。如果你向后看代码，生成 ID 的核心函数在 kombu.utils.uuid.uuid(..) 中，有趣的是它只是内置 uuid.uuid4(..) 的一个薄包装。

但是，如果您查看函数签名：

def uuid(_uuid=uuid4):
    """Generate unique id in UUID4 format.
    See Also:
        For now this is provided by :func:`uuid.uuid4`.
    """
    return str(_uuid())

..似乎可以提供您自己的生成 ID 的函数，只要它们是唯一的，它们就可以工作。从根本上说，您必须修补：celery.__init__.uuid、celery.utils.__init__.uuid 和 celery.utils.__init__.gen_unique_id。

我认为您不能应用有价值的名称，因为调用此函数时不带任何参数来简单地返回一些独特的东西。

但是

如果您查看Task.apply 和Task.apply_async，则有一个未记录的参数task_id！

这将允许您在调用/创建任务之前手动指定task_id，只要它们是全球唯一的，您仍然可以获得所有报告和指标。不幸的是，我没有看到任何简单的方法可以从结果中更改 task_id，以使其在事后更有用......有点让你的 UI 的文本框有点傻。

---

如果您的项目是为它设置的，您可以使用任务继承并以这种方式修改任务行为。理论上你应该可以覆盖apply 和apply_async 来注入任务的ID。

import time

from celery import Task

 class NamedTask(Task):
    id_gen = lambda: int(time.time() * 1000)


    def _gen_task_id(self):
        return 
            'task_id': '%s-%s' % (
                self.name,
                self.id_gen())

    def apply(self, *args, **kwargs):
        kwargs.update(self._gen_task_id())
        return Task.apply(self, *args, **kwargs)

    def apply_async(self, *args, **kwargs):
        kwargs.update(self._gen_task_id())
        return Task.apply_async(self, *args, **kwargs)



@task(base=NamedTask)
def add(x, y):
    return x + y

应该让你的任务 ID 看起来像：project.tasks.add-15073315189870

【讨论】：

您好，我不想更改任务 ID，这很好，我想知道是否可以向模型添加另一个字段，在其中我可以对其进行描述并在 django_celery_resutls 数据库视图中查看？

@AlexW，你的问题应该已经很清楚了。因为现在问题本身的含义发生了变化。你的标题清楚地写着set task id to something human readable?。

为混乱道歉，重新阅读我可以看到问题需要改进

【参考方案2】：

实现一个自定义结果后端，它会覆盖_store_result 来决定将什么作为结果保存到数据库中。

根据您使用的后端，在 celery.backends 中查找相关类。

此示例扩展了 amqp 后端的结果。

class UsefulInfoBackend(AMQPBackend):
    def store_result(self, task_id, result, state,
                     traceback=None, request=None, **kwargs):
        result = super(UsefulInfoBackend, self).store_result(task_id, result, state,
                     traceback=None, request=None, **kwargs)
        result['useful_info'] = 'Very Useful! :)' # determine the rules for extraneous information here contains. 
        return result

在初始化 Celery 时，传递你的结果后端类。

celery.Celery(backend=UsefulInfoBackend)

【参考方案3】：

试试：

@app.task(name="periodic_test")
def test(a, b):
    return a+b

这可能有助于在 UI 中查找任务状态，但 Celery 总是需要每个任务执行的唯一 ID。