Using Celery Signals
Celery provides a signaling system that allows you to hook into various stages of a task's lifecycle. This is useful for logging, monitoring, or triggering side effects (like sending an email notification) when a task starts, succeeds, or fails.
How Signals Work
Signals use the Observer pattern. You define a function (the receiver) and connect it to a specific signal. When that event occurs in the worker, your function is executed.
Important: Signal handlers run synchronously within the worker process. If your signal handler is slow (e.g., making a network request), it will block the worker from processing other tasks.
Common Signals
task_prerun: Sent before a task is executed.task_postrun: Sent after a task is executed (regardless of success or failure).task_success: Sent when a task succeeds.task_failure: Sent when a task fails.worker_ready: Sent when the worker has started.
Connecting to Signals
The easiest way to connect to a signal is using the @connect decorator provided by the signal object.
Example 1: Logging Task Lifecycle
from celery.signals import task_prerun, task_postrun
from celery.utils.log import get_task_logger
logger = get_task_logger(__name__)
@task_prerun.connect
def task_starting_handler(sender=None, headers=None, body=None, **kwargs):
# sender is the task object
info = headers if headers else {}
logger.info(f"Task {sender.name} is starting. ID: {info.get('id')}")
@task_postrun.connect
def task_finished_handler(sender=None, task_id=None, state=None, **kwargs):
logger.info(f"Task {task_id} finished with state: {state}")Example 2: Handling Task Failure
This is a common use case for sending alerts to Slack, Email, or Sentry when a critical task fails.
from celery.signals import task_failure
@task_failure.connect
def handle_task_failure(sender=None, task_id=None, exception=None, traceback=None, **kwargs):
print(f"CRITICAL: Task {task_id} failed!")
print(f"Exception: {exception}")
# Example: Send alert (ensure this is fast or async!)
# send_slack_alert(f"Task {sender.name} failed: {exception}")Filtering Signals
You can restrict a signal handler to specific tasks by checking the sender argument, which usually contains the task object or task name.
@task_success.connect(sender='tasks.add')
def add_success_handler(sender=None, result=None, **kwargs):
print(f"The add task finished with result: {result}")