如何获取计划的芹菜任务的任务对象？答案

【问题标题】：How to get Task objects for scheduled celery tasks?如何获取计划的芹菜任务的任务对象？
【发布时间】：2019-12-09 13:35:13
【问题描述】：

这个问题类似于Retrieve list of tasks in a queue in Celery，但我想获取实际的Task 对象（参见http://docs.celeryproject.org/en/latest/reference/celery.events.state.html#celery.events.state.Task）而不是字典表示。

当我这样做时

from celery.task.control import inspect
i = inspect()

然后从 shell 执行i.scheduled()，我得到类似的结果

In [75]: i.scheduled()
Out[75]: 
{'celery@Kurts-MacBook-Pro-3.local': [{'eta': '2019-08-01T01:31:37.843141+00:00',
   'priority': 6,
   'request': {'acknowledged': False,
    'args': "[126, 'Business Signup', {'actualCompanyName': 'Gilmore and Beck LLC', 'gigsEnabledRegionMapping': True, 'companyName': 'Gilmore and Beck LLC', 'companyRegionMapping': 'Lesliechester', 'companyId': 'ow9DMA8'}]",
    'delivery_info': {'exchange': '',
     'priority': 0,
     'redelivered': None,
     'routing_key': 'celery'},
    'hostname': 'celery@Kurts-MacBook-Pro-3.local',
    'id': '4ecdc400-8421-4a06-babc-98493362ec67',
    'kwargs': '{}',
    'name': 'backend.tasks.task_send_event_to_iterable',
    'time_start': None,
    'type': 'backend.tasks.task_send_event_to_iterable',
    'worker_pid': None}},
  {'eta': '2019-08-01T01:39:21.205879+00:00',
   'priority': 6,
   'request': {'acknowledged': False,
    'args': "('hyjomuz@mailinator.net',)",
    'delivery_info': {'exchange': '',
     'priority': 0,
     'redelivered': None,
     'routing_key': 'celery'},
    'hostname': 'celery@Kurts-MacBook-Pro-3.local',
    'id': '294910a3-2323-4fcf-9768-115c1a8c5e06',
    'kwargs': '{}',
    'name': 'backend.tasks.task_send_business_lead_notification',
    'time_start': None,
    'type': 'backend.tasks.task_send_business_lead_notification',
    'worker_pid': None}}]}

我想对这些任务进行搜索并有条件地撤销一项。但是，我想迭代结果并方便地执行实际任务，就像How to inspect and cancel Celery tasks by task name 中的示例一样。但是如果我尝试做celery.events.state.State()，我不会收到任何事件：

In [76]: celery.events.state.State()
Out[76]: <State: events=0 tasks=0>

是否可以为计划任务获取实际的Task 对象？特别是，我有兴趣获得args 而无需解析或尝试ast.literal_eval()。

【问题讨论】：

标签： python celery

【解决方案1】：

我不确定您是否能真正获得实际的 Task 实例，但您可以通过简单地使用您要检查的任务 ID 实例化它来轻松创建 AsyncResult，并且您需要（自然）将 Celery 应用程序对象传递给也是。

一些伪代码：

from celery.result import AsyncResult
from my.project.celeryapp import myapp

task_res = AsyncResult("9ed888fe-f6b6-4443-85d3-787c5c1b26b0", app=myapp)
print(task_res.state)

【讨论】：