Celery Custom Task Classes: Building Reusable Task Base Classes with Shared Behavior

In this tutorial, you will learn about Celery Custom Task Classes: Building Reusable Task Base Classes with Shared Behavior. We cover key concepts, practical examples, and best practices to help you master this topic.

Celery custom task classes let you define a base class that all tasks inherit from, enabling shared behavior like automatic monitoring, database session management, metrics emission, error tracking, and consistent retry policies across your entire task collection.

flowchart TD
    Base[BaseTask Class] --> T1[Task A]
    Base --> T2[Task B]
    Base --> T3[Task C]
    Base -.->|Shared| L[Logging]
    Base -.->|Shared| M[Metrics]
    Base -.->|Shared| DB[DB Sessions]
    Base -.->|Shared| R[Retry Policy]
    Base -.->|Shared| E[Error Tracking]

What You'll Learn

• Creating custom base task classes
• Overriding on_success, on_failure, on_retry
• Adding database session management
• Automatic metrics and monitoring
• Custom retry policies per task group

Why It Matters

Without custom task classes, shared logic (logging, DB cleanup, metrics) must be duplicated in every task. A custom base class centralizes cross-cutting concerns, reduces boilerplate, ensures consistent error handling, and simplifies adding new capabilities across all tasks.

Real-World Use

DodaTech's platform has a BaseAnalysisTask that handles database session cleanup and sends completion metrics to Prometheus. Each of the 50+ analysis tasks inherits from this base, getting monitoring for free. The operations team sees per-task latency and error rates on a single dashboard.

Custom Base Task

from celery import Celery, Task
import time
import json

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

class MonitoredTask(Task):
    abstract = True

    def on_success(self, retval, task_id, args, kwargs):
        duration = time.time() - self.request.timestamp
        print(f"[MONITOR] Task {self.name} succeeded")
        print(f"[MONITOR]   task_id: {task_id}")
        print(f"[MONITOR]   duration: {duration:.3f}s")
        print(f"[MONITOR]   args: {args}")

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        print(f"[MONITOR] Task {self.name} FAILED")
        print(f"[MONITOR]   task_id: {task_id}")
        print(f"[MONITOR]   error: {exc}")
        print(f"[MONITOR]   traceback: {einfo}")

    def on_retry(self, exc, task_id, args, kwargs, einfo):
        retry_num = self.request.retries
        print(f"[MONITOR] Task {self.name} retry #{retry_num}")
        print(f"[MONITOR]   error: {exc}")

@app.task(base=MonitoredTask)
def process_file(file_path):
    print(f"Processing {file_path}")
    time.sleep(0.1)
    return f"{file_path} processed"

@app.task(base=MonitoredTask)
def api_call(endpoint, payload):
    print(f"Calling {endpoint}")
    time.sleep(0.05)
    return f"{endpoint} responded"

task1 = process_file.delay("/data/file1.csv")
task2 = api_call.delay("/api/users", {"id": 42})
print(f"Tasks submitted: {task1.id}, {task2.id}")

Expected output:

Tasks submitted: id1, id2
Processing /data/file1.csv
[MONITOR] Task custom.process_file succeeded
[MONITOR]   task_id: id1
[MONITOR]   duration: 0.102s
Calling /api/users
[MONITOR] Task custom.api_call succeeded
[MONITOR]   task_id: id2
[MONITOR]   duration: 0.052s

Database Session Task

from celery import Celery, Task
import time

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

class DatabaseTask(Task):
    abstract = True
    _db_connections = {}

    def before_start(self, task_id, args, kwargs):
        print(f"[DB] Opening database connection for {self.name}")
        self._db_connections[task_id] = {
            'connected': True,
            'started_at': time.time(),
        }

    def after_return(self, status, retval, task_id, args, kwargs, einfo):
        conn = self._db_connections.pop(task_id, None)
        if conn:
            elapsed = time.time() - conn['started_at']
            print(f"[DB] Closing database connection for {self.name}")
            print(f"[DB]   duration: {elapsed:.3f}s")
            print(f"[DB]   status: {status}")

    def run_db_query(self, query):
        print(f"[DB] Executing: {query}")
        time.sleep(0.2)
        return f"Results for: {query}"

@app.task(base=DatabaseTask)
def get_user_data(user_id):
    task = get_user_data
    task.run_db_query(f"SELECT * FROM users WHERE id = {user_id}")
    time.sleep(0.1)
    return {"user_id": user_id, "name": "Alice"}

result = get_user_data.delay(42)
print(f"User data task: {result.id}")

Expected output:

User data task: id
[DB] Opening database connection for custom.get_user_data
[DB] Executing: SELECT * FROM users WHERE id = 42
[DB] Closing database connection for custom.get_user_data
[DB]   duration: 0.305s
[DB]   status: SUCCESS

Custom Retry Task

from celery import Celery, Task
import random

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

class RetryableTask(Task):
    abstract = True
    max_retries = 5
    default_retry_delay = 10

    def on_retry(self, exc, task_id, args, kwargs, einfo):
        print(f"[RETRY] {self.name} attempt {self.request.retries + 1}/{self.max_retries}")
        print(f"[RETRY]   waiting {self.request.retry_delay}s")
        print(f"[RETRY]   error: {exc}")

    def before_retry(self, exc, task_id, args, kwargs, einfo):
        if self.request.retries > self.max_retries:
            print(f"[RETRY] Max retries reached for {self.name}, failing permanently")
        return super().before_retry(exc, task_id, args, kwargs, einfo)

@app.task(base=RetryableTask, bind=True, max_retries=3)
def unreliable_api(self, endpoint):
    if random.random() < 0.7:
        raise ConnectionError(f"Failed to reach {endpoint}")
    result = f"Called {endpoint} successfully"
    print(result)
    return result

task = unreliable_api.delay("https://api.example.com/data")
print(f"API call: {task.id}")

Expected output:

API call: id
[RETRY] custom.unreliable_api attempt 1/3
[RETRY]   waiting 10s
[RETRY]   error: Failed to reach https://api.example.com/data
[RETRY] custom.unreliable_api attempt 2/3
[RETRY]   waiting 10s
Called https://api.example.com/data successfully

Common Mistakes

• Forgetting abstract=True on the base class -- without abstract=True, Celery registers the base class as a regular task. If you try to call it directly, it fails. Always set abstract=True on base task classes.
• Not calling super() in overridden methods -- overriding on_success without super() bypasses the default task completion logic. Call super() if you want to preserve standard behavior.
• Modifying self.request in before_start -- the request object should be treated as read-only. Modifying it in hooks can cause unpredictable state propagation across the task lifecycle.
• Heavy operations in hooks -- hooks run synchronously in the worker Process. Long-running hooks block task execution. Keep hooks fast (log, emit metric, clean up) and avoid external calls.
• Circular imports with task classes -- base task classes often need imports from the task module, causing circular imports. Define base classes in a separate module that does not import task modules.

Practice Questions

1. Why set abstract=True on a custom task base class?
2. What lifecycle hooks are available in Celery task classes?
3. How do you pass the task instance to a custom base class method?
4. What happens if on_failure raises an exception?
5. How can you share state across tasks using a custom base class?

Challenge

Build a comprehensive base task class that: (1) opens and closes database sessions automatically (before_start/after_return), (2) tracks execution time and emits timing metrics via a callback, (3) implements exponential backoff retry with jitter, (4) logs all task results and failures to a task history table, (5) sends a heartbeat to Redis every 30 seconds of execution to detect stuck tasks, and (6) provides a decorator @requires_db and @requires_redis for selective resource management.

FAQ

Mini Project

Build a production-ready base task framework: (1) BaseTask: automatic error logging to Sentry/Prometheus, execution time tracking, task timeout warning, (2) DatabaseTask(BaseTask): automatic SQLAlchemy session lifecycle, Transaction management, connection leak detection, (3) RetryTask(BaseTask): exponential backoff with jitter, max retries per task group, dead letter handling, (4) BatchTask(DatabaseTask): chunked processing with progress tracking, checkpoint save, resume from failure. Include unit tests for each base class.

What's Next

Continue with Unit Testing to learn how to test tasks with custom base classes. Then explore Docker Deployment for containerized Celery deployments.