check if task is scheduled/executing with specifc arguments?

I'm using Celery with Redis. Is there a way to check if a certain task is scheduled (or executing) with specific arguments?

I know I can do this by getting the task id when scheduling a task with apply_async and later checking with AsyncResult. But I'm wondering if there is a way to inspect the queues to find if it has been scheduled without knowing the task id.

Yes, that is possible with the monitoring and management API.

To get a list of scheduled tasks:

celery -A proj inspect scheduled

To get a list of currently executing (active) tasks:

celery -A proj inspect active

Your application module is probably not proj like in these examples above - you'll need to adjust that.

Also, keep in mind inspect can be executed programmatically via the celery.app.control.Inspect class. I use this quite a lot in my Celery monitoring scripts.

from proj.celery import app

inspect = app.control.inspect()

active_tasks = inspect.active()
reserved_tasks = inspect.reserved()
scheduled_tasks = inspect.scheduled()

if active_tasks:
    for worker_name, tasks in active_tasks.items():
        for task in tasks:
            print(task['name'], task['args'], task['kwargs'])