(task_names, interval=5)
Constantly polls all workers for any active, reserved, or scheduled
    tasks, and revokes the tasks if they match any of the given task names.

    Note that when reserved or scheduled tasks are revoked, they don't disappear from
    the list of reserved or scheduled tasks immediately. Instead, they will just be
    discarded by the worker when they try to move to active.

    :param task_names: a list of fully qualified task names to revoke
    :param interval: the interval (in seconds) on which to poll the workers for tasks

    Example:
    revoke_tasks(['couchexport.tasks.export_async'])
Constantly polls all workers for any active, reserved, or scheduled
    tasks, and revokes the tasks if they match any of the given task names.
Constantly polls all workers for any active reserved or scheduled tasks and revokes the tasks if they match any of the given task names .
def revoke_tasks(task_names, interval=5):
    """
    Constantly polls all workers for any active, reserved, or scheduled
    tasks, and revokes the tasks if they match any of the given task names.

    Note that when reserved or scheduled tasks are revoked, they don't disappear from
    the list of reserved or scheduled tasks immediately. Instead, they will just be
    discarded by the worker when they try to move to active.

    :param task_names: a list of fully qualified task names to revoke
    :param interval: the interval (in seconds) on which to poll the workers for tasks

    Example:
    revoke_tasks(['couchexport.tasks.export_async'])
    """
    app = Celery()
    app.config_from_object(settings)
    task_ids = set()
    while True:
        tasks = []
        inspect = app.control.inspect()
        for task_state in ['active', 'reserved', 'scheduled']:
            result = getattr(inspect, task_state)()
            if not result:
                continue

            for worker, task_dicts in result.items():
                tasks.extend(_get_task_info_fcn(task_state)(task_dicts))

        for task in tasks:
            if task.name in task_names and task.id not in task_ids:
                app.control.revoke(task.id, terminate=True)
                task_ids.add(task.id)
                print(datetime.utcnow(), 'Revoked', task.id, task.name)

        sleep(interval)
def revoke_tasks ( task_names , interval = 5 ) : app = Celery ( ) app . config_from_object ( settings ) task_ids = set ( ) while True : tasks = [ ] inspect = app . control . inspect ( ) for task_state in [ 'active' , 'reserved' , 'scheduled' ] : result = getattr ( inspect , task_state ) ( ) if not result : continue for worker , task_dicts in result . items ( ) : tasks . extend ( _get_task_info_fcn ( task_state ) ( task_dicts ) ) for task in tasks : if task . name in task_names and task . id not in task_ids : app . control . revoke ( task . id , terminate = True ) task_ids . add ( task . id ) print ( datetime . utcnow ( ) , 'Revoked' , task . id , task . name ) sleep ( interval )