Celery中如何禁用task_success信号的触发调用？

Problem Context #

I'm using Celery signals successfully with the following code:

@shared_task(name="execute_task")
def execute(*args,**kwargs):
    return 2+2

@task_success.connect
def task_success_handler(sender=None, headers=None, body=None, **kwargs):
    print(pass)  # Note: This is invalid syntax, likely a typo

When the execute_task completes, the task_success signal triggers task_success_handler, and I want to disable this specific trigger. Is this possible?

Solution #

Absolutely, you can disable the signal handler either entirely or just for your execute_task — here are two practical approaches:

1. Disable the Handler Entirely #

If you want to stop this handler from running for all tasks, use Celery's signal disconnect method. Add this line in your app initialization code (e.g., where you set up your Celery instance):

from celery.signals import task_success

# Disconnect the handler so it never fires
task_success.disconnect(task_success_handler)

Once executed, the task_success_handler won't be triggered by any task completion event.

2. Disable Only for execute_task #

If you want the handler to work for other tasks but skip when execute_task finishes, modify the handler to check the sender task's identity before executing your logic:

@task_success.connect
def task_success_handler(sender=None, headers=None, body=None, **kwargs):
    # Exit early if the completed task is execute_task
    if sender and sender.name == "execute_task":
        return
    
    # Replace with your actual logic (fixing the invalid print(pass) line)
    print(f"Task {sender.name} completed successfully!")

This way, the handler will still run for other tasks, but it'll skip execution specifically when execute_task finishes.

Bonus: Temporary Dynamic Disable (For Specific Runs) #

If you only need to disable the handler for a single execution of execute_task, you can connect/disconnect dynamically around the task call:

# Before calling the task, disconnect the handler
task_success.disconnect(task_success_handler)

# Execute the task
execute.delay()

# Reconnect the handler if needed
task_success.connect(task_success_handler)

This is useful for one-off scenarios where you don't want the handler to run for a specific task instance.

内容的提问来源于stack exchange，提问作者Ravinder Baid