Celery Signals: Extending Task Behavior with Lifecycle Hooks

In this tutorial, you will learn about Celery Signals: Extending Task Behavior with Lifecycle Hooks. We cover key concepts, practical examples, and best practices to help you master this topic.

Celery signals are Django-style event hooks that let you execute custom code at specific points in the task and worker lifecycle — before a task is published, before it runs, after it succeeds or fails — enabling cross-cutting concerns like logging, metrics, and audit trails.

flowchart LR
    Publish[Publish Task] -->|before_task_publish| PrePublish[Log Publish]
    PrePublish --> Worker[Worker Receives]
    Worker -->|task_prerun| PreRun[Setup / DB Connect]
    PreRun --> Execute[Execute Task]
    Execute -->|task_success| Success[Log Success / Metrics]
    Execute -->|task_failure| Failure[Log Error / Alert]
    Execute -->|task_postrun| PostRun[Cleanup / Close DB]

What You'll Learn

• Signal registration and receiver functions
• Built-in Celery signals: before_task_publish, task_prerun, task_postrun, task_success, task_failure
• Cross-cutting concerns: logging, metrics, audit
• Signal ordering and error handling in signals

Why It Matters

Without signals, adding cross-cutting behavior like logging every task execution or tracking metrics requires modifying every task function. Signals let you add this behavior once, centrally, without touching individual task code. This keeps tasks clean and focused on business logic.

Real-World Use

DodaTech's Celery signals: a task_prerun signal opens a database connection and traces task start in OpenTelemetry. A task_success signal records duration and closes the span. A task_failure signal sends the exception to Sentry. All tasks get this behavior automatically — no task function was modified.

Task Lifecycle Signals

Register signal handlers for task lifecycle events:

from celery import Celery, signals
import time

app = Celery('signals_demo', broker='redis://localhost:6379/0')

@signals.before_task_publish.connect
def before_publish(sender=None, headers=None, body=None, **kwargs):
    task_id = headers.get('id', 'unknown')
    task_name = headers.get('task', 'unknown')
    print(f"[SIGNAL] Publishing task {task_name} ({task_id})")

@signals.task_prerun.connect
def task_prerun(sender=None, task_id=None, task=None, **kwargs):
    print(f"[SIGNAL] Task {task_id} starting at {time.time():.2f}")

@signals.task_postrun.connect
def task_postrun(sender=None, task_id=None, task=None, **kwargs):
    print(f"[SIGNAL] Task {task_id} finished at {time.time():.2f}")

@signals.task_success.connect
def task_success(sender=None, result=None, **kwargs):
    print(f"[SIGNAL] Task succeeded with result: {result}")

@signals.task_failure.connect
def task_failure(sender=None, task_id=None, exception=None, **kwargs):
    print(f"[SIGNAL] Task {task_id} failed: {exception}")

@app.task
def successful_task(name):
    time.sleep(0.1)
    return f"Hello {name}"

@app.task
def failing_task():
    time.sleep(0.1)
    raise ValueError("Something went wrong")

successful_task.delay("Alice")
time.sleep(0.3)

failing_task.delay()
time.sleep(0.3)

Expected output:

[SIGNAL] Publishing task signals_demo.successful_task (550e8400-...)
[SIGNAL] Task 550e8400-... starting at 1719580800.12
[SIGNAL] Task 550e8400-... finished at 1719580800.22
[SIGNAL] Task succeeded with result: Hello Alice
[SIGNAL] Publishing task signals_demo.failing_task (550e8401-...)
[SIGNAL] Task 550e8401-... starting at 1719580800.45
[SIGNAL] Task 550e8401-... finished at 1719580800.55
[SIGNAL] Task 550e8401-... failed: Something went wrong

Metrics and Monitoring Signals

Track task metrics using signals:

from celery import Celery, signals
import time
from collections import defaultdict

app = Celery('signals_demo', broker='redis://localhost:6379/0')

class TaskMetrics:
    def __init__(self):
        self.task_durations = defaultdict(list)
        self.task_counts = defaultdict(int)
        self.task_errors = defaultdict(int)
        self.active_tasks = {}

    def on_prerun(self, task_id, task_name, **kwargs):
        self.active_tasks[task_id] = {"name": task_name, "start": time.time()}

    def on_success(self, task_id, result, **kwargs):
        info = self.active_tasks.pop(task_id, {})
        name = info.get("name", "unknown")
        duration = time.time() - info.get("start", time.time())
        self.task_durations[name].append(duration)
        self.task_counts[name] += 1
        self._log_metric(name, "success", duration)

    def on_failure(self, task_id, exception, **kwargs):
        info = self.active_tasks.pop(task_id, {})
        name = info.get("name", "unknown")
        self.task_counts[name] += 1
        self.task_errors[name] += 1
        self._log_metric(name, "failure", 0)

    def _log_metric(self, task_name, status, duration):
        durations = self.task_durations[task_name]
        recent = durations[-10:] if len(durations) >= 10 else durations
        avg = sum(recent) / len(recent)
        print(f"[METRIC] {task_name:40s} {status:8s} "
              f"count={self.task_counts[task_name]:3d} "
              f"errors={self.task_errors[task_name]:2d} "
              f"avg_duration={avg:.3f}s")

    def report(self):
        print("\n=== Task Metrics Report ===")
        for name, durations in self.task_durations.items():
            avg = sum(durations) / len(durations)
            print(f"{name:40s} count={self.task_counts[name]:4d} "
                  f"errors={self.task_errors[name]:3d} "
                  f"avg={avg:.3f}s "
                  f"total={sum(durations):.2f}s")

metrics = TaskMetrics()

@signals.task_prerun.connect
def metrics_prerun(sender=None, task_id=None, task=None, **kwargs):
    metrics.on_prerun(task_id, task.name)

@signals.task_success.connect
def metrics_success(sender=None, task_id=None, result=None, **kwargs):
    metrics.on_success(task_id, result)

@signals.task_failure.connect
def metrics_failure(sender=None, task_id=None, exception=None, **kwargs):
    metrics.on_failure(task_id, exception)

@app.task
def quick_task():
    return "quick"

@app.task
def slow_task():
    time.sleep(0.2)
    return "slow"

for _ in range(3):
    quick_task.delay()
    slow_task.delay()

time.sleep(2)
metrics.report()

Expected output:

[METRIC] signals_demo.quick_task                         success  count=  1 errors= 0 avg_duration=0.002s
[METRIC] signals_demo.slow_task                          success  count=  1 errors= 0 avg_duration=0.202s
...

=== Task Metrics Report ===
signals_demo.quick_task                       count=   3 errors=   0 avg=0.002s total=0.01s
signals_demo.slow_task                        count=   3 errors=   0 avg=0.202s total=0.61s

Common Mistakes

• Performing slow operations in signal handlers — signals run in the same Process as the task. A slow signal handler delays the task. Keep signal handlers fast. Use async operations or background threads for heavy processing.
• Modifying task arguments in before_task_publish — while you can modify headers/body, this can cause unexpected behavior. Prefer task decorator options over signal-based argument modification.
• Not handling exceptions in signal handlers — an exception in a signal handler propagates and can prevent the task from executing. Wrap signal handler code in try/except blocks.
• Registering the same signal multiple times — duplicate registrations cause duplicate execution. Use weak=False or check for existing connections to prevent duplicates.
• Assuming signal ordering — the order of multiple signal handlers for the same event is not guaranteed. If order matters, use a single handler that calls sub-functions in order.

Practice Questions

1. How do Celery signals differ from the Celery event system?
2. What is the purpose of the before_task_publish signal?
3. Why should signal handlers be fast and non-blocking?
4. How do you add custom data to task signals using headers?
5. What happens when a signal handler raises an exception?

Challenge

Build a Celery signal-based audit system that: (1) logs every task publish with caller info (module, line number), (2) tracks task execution duration and stores in a time-series database, (3) captures task return values for debugging (for tasks marked with a @audit decorator), (4) sends Slack notifications for task failures exceeding a threshold (3 failures in 5 minutes of the same task type), and (5) provides a signal handler that auto-retries failed tasks with modified arguments (e.g., longer timeout).

FAQ

Mini Project

Build a Celery signal-based cross-cutting concern library that provides: (1) automatic OpenTelemetry tracing for every task (prerun = start span, postrun = end span), (2) Prometheus metrics for task duration, count, and error rate, (3) structured logging with task ID, name, and duration, (4) database connection pooling with automatic checkout/checkin around task execution, and (5) dead-letter handling that routes failed tasks to a secondary queue. All via signal handlers — no task code changes needed.

What's Next

Continue with Celery Logging to learn about structured logging for Celery tasks. Then explore Celery Serialization for task argument Serialization options.