I am trying to find a way to delete all the currently queued tasks with a specific given name from a Celery queue. From the official documentations, I know I could inspect the workers and revoke the tasks by looking for their name and then getting their IDs like:
def drop_celery_task(options):
def _get_tasks_id(workers: list, tasks_ids: list, task_name: str):
"""
Get task ids with the given name included inside the given `workers` tasks.
{'worker1.example.com': [
{'name': 'tasks.sleeptask', 'id': '32666e9b-809c-41fa-8e93-5ae0c80afbbf',
'args': '(8,)', 'kwargs': '{}'}]
}
"""
for worker in workers:
if not workers[worker]:
continue
for _task in workers[worker]:
if _task["name"].split(".")[-1] == task_name:
tasks_ids.append(_task["id"])
task_name = options.drop_celery_task["name"]
i = Inspect(app=celery_app) # Inspect all nodes.
registered = i.registered()
if not registered:
raise Exception("No registered tasks found")
if not any(task_name == worker.split(".")[-1] for worker in chain(*list(registered.values()))):
raise Exception(f"Task not registered: {task_name}")
tasks_ids = []
_get_tasks_id(i.active(), tasks_ids, task_name)
_get_tasks_id(i.scheduled(), tasks_ids, task_name)
_get_tasks_id(i.reserved(), tasks_ids, task_name)
if tasks_ids:
for task_id in tasks_ids:
Control(app=celery_app).revoke(task_id)
else:
logging.info(f"No active/scheduled/registered task found with the name {task_name}")
But this code only revokes the tasks fetched or pre-fetched by the celery workers, not the ones still in the queue (using Redis as backend). Any advice on how to remove the ones in Redis using celery commands, or prevent the workers from accepting tasks with a given name?