django-celery

Django + Celery async task patterns — configuration, task design, beat scheduling, retries, canvas workflows, monitoring, and testing. Use when adding…

view source

installs

stars

karma

view original SKILL.md from skills.shclick to expand

Django + Celery Async Task Patterns

Production-grade patterns for background task processing in Django using Celery with Redis or RabbitMQ.

When to Activate

Adding background jobs or async processing to a Django app

Implementing periodic/scheduled tasks

Offloading slow operations (email, PDF generation, API calls) from request cycle

Setting up Celery Beat for cron-like scheduling

Debugging task failures, retries, or queue backlogs

Writing tests for Celery tasks

Project Setup

Installation
pip install 'celery[redis]' django-celery-results django-celery-beat

celery.py — App Entrypoint

# config/celery.py
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.development')

app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()  # Discovers tasks.py in each INSTALLED_APP

@app.task(bind=True, ignore_result=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

# config/__init__.py
from .celery import app as celery_app

__all__ = ('celery_app',)

Django Settings

# config/settings/base.py

# Broker (Redis recommended for production)
CELERY_BROKER_URL = env('CELERY_BROKER_URL', default='redis://localhost:6379/0')
CELERY_RESULT_BACKEND = env('CELERY_RESULT_BACKEND', default='django-db')

# Serialization
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

# Task behavior
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = 30 * 60        # Hard limit: 30 min
CELERY_TASK_SOFT_TIME_LIMIT = 25 * 60   # Soft limit: sends SoftTimeLimitExceeded
CELERY_WORKER_PREFETCH_MULTIPLIER = 1   # Prevent worker hoarding long tasks
CELERY_TASK_ACKS_LATE = True            # Re-queue on worker crash

# Result persistence
CELERY_RESULT_EXPIRES = 60 * 60 * 24   # Keep results 24 hours

# Beat scheduler (for periodic tasks)
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

# Installed apps
INSTALLED_APPS += [
    'django_celery_results',
    'django_celery_beat',
]

Running Workers

# Start worker (development)
celery -A config worker --loglevel=info

# Start beat scheduler (periodic tasks)
celery -A config beat --loglevel=info --scheduler django_celery_beat.schedulers:DatabaseScheduler

# Combined worker + beat (dev only, never production)
celery -A config worker --beat --loglevel=info

# Production: multiple workers with concurrency
celery -A config worker --loglevel=warning --concurrency=4 -Q default,high_priority

Task Design Patterns

Basic Task

# apps/notifications/tasks.py
from celery import shared_task
import logging

logger = logging.getLogger(__name__)

@shared_task(name='notifications.send_welcome_email')
def send_welcome_email(user_id: int) -> None:
    """Send welcome email to newly registered user."""
    from apps.users.models import User
    from apps.notifications.services import EmailService

    try:
        user = User.objects.get(pk=user_id)
    except User.DoesNotExist:
        logger.warning('send_welcome_email: user %s not found', user_id)
        return  # Idempotent — do not raise, task already impossible to complete

    EmailService.send_welcome(user)
    logger.info('Welcome email sent to user %s', user_id)

Retryable Task

@shared_task(
    bind=True,
    name='integrations.sync_to_crm',
    max_retries=5,
    default_retry_delay=60,       # seconds before first retry
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,           # exponential backoff
    retry_backoff_max=600,        # cap at 10 minutes
    retry_jitter=True,            # randomise to avoid thundering herd
)
def sync_contact_to_crm(self, contact_id: int) -> dict:
    """Sync contact to external CRM with retry on transient failures."""
    from apps.crm.services import CRMClient

    try:
        result = CRMClient().sync(contact_id)
        return result
    except CRMClient.RateLimitError as exc:
        # Specific retry delay from response header
        raise self.retry(exc=exc, countdown=int(exc.retry_after))

Idempotent Task Pattern

Design tasks so they can safely run multiple times with the same inputs:

@shared_task(name='orders.mark_shipped')
def mark_order_shipped(order_id: int, tracking_number: str) -> None:
    """Mark order as shipped — safe to run multiple times."""
    from apps.orders.models import Order

    updated = Order.objects.filter(
        pk=order_id,
        status=Order.Status.PROCESSING,    # Guard: only update if not already shipped
    ).update(
        status=Order.Status.SHIPPED,
        tracking_number=tracking_number,
    )

    if not updated:
        logger.info('mark_order_shipped: order %s already shipped or not found', order_id)

Task with Soft Time Limit

from celery.exceptions import SoftTimeLimitExceeded

@shared_task(
    bind=True,
    name='reports.generate_pdf',
    soft_time_limit=120,
    time_limit=150,
)
def generate_pdf_report(self, report_id: int) -> str:
    """Generate PDF report with graceful timeout handling."""
    from apps.reports.services import PDFGenerator

    try:
        path = PDFGenerator.build(report_id)
        return path
    except SoftTimeLimitExceeded:
        # Clean up partial files before hard kill
        PDFGenerator.cleanup(report_id)
        raise

Calling Tasks

from datetime import timedelta
from django.utils import timezone

# Fire and forget (async)
send_welcome_email.delay(user.pk)

# Schedule in the future
send_reminder.apply_async(args=[user.pk], countdown=3600)  # 1 hour from now
send_reminder.apply_async(args=[user.pk], eta=timezone.now() + timedelta(days=1))

# Apply with queue routing
sync_contact_to_crm.apply_async(args=[contact.pk], queue='high_priority')

# Run synchronously (tests / debugging only)
result = generate_pdf_report.apply(args=[report.pk])

Beat Scheduling (Periodic Tasks)

Code-Defined Schedule

# config/settings/base.py
from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    'cleanup-expired-sessions': {
        'task': 'users.cleanup_expired_sessions',
        'schedule': crontab(hour=2, minute=0),   # 2am daily
    },
    'sync-inventory': {
        'task': 'products.sync_inventory',
        'schedule': 60.0,                         # every 60 seconds
    },
    'weekly-digest': {
        'task': 'notifications.send_weekly_digest',
        'schedule': crontab(day_of_week='monday', hour=8, minute=0),
    },
}

Database-Defined Schedule (via django-celery-beat)

# Manage periodic tasks from Django admin or code
from django_celery_beat.models import PeriodicTask, CrontabSchedule
import json

schedule, _ = CrontabSchedule.objects.get_or_create(
    hour='*/6', minute='0',
    timezone='UTC',
)

PeriodicTask.objects.update_or_create(
    name='Sync inventory every 6 hours',
    defaults={
        'crontab': schedule,
        'task': 'products.sync_inventory',
        'args': json.dumps([]),
        'enabled': True,
    }
)

Canvas: Chaining and Grouping Tasks

from celery import chain, group, chord

# Chain: run tasks sequentially, passing results
pipeline = chain(
    fetch_data.s(source_id),
    transform_data.s(),          # receives fetch_data result as first arg
    load_to_warehouse.s(),
)
pipeline.delay()

# Group: run tasks in parallel
parallel = group(
    send_welcome_email.s(user_id)
    for user_id in new_user_ids
)
parallel.delay()

# Chord: parallel tasks + callback when all complete
result = chord(
    group(process_chunk.s(chunk) for chunk in data_chunks),
    aggregate_results.s(),       # called with list of chunk results
)
result.delay()

Error Handling and Dead Letter Queue

# apps/core/tasks.py
from celery.signals import task_failure

@task_failure.connect
def on_task_failure(sender, task_id, exception, args, kwargs, traceback, einfo, **kw):
    """Log all task failures to Sentry / alerting."""
    import sentry_sdk
    with sentry_sdk.new_scope() as scope:
        scope.set_context('celery', {
            'task': sender.name,
            'task_id': task_id,
            'args': args,
            'kwargs': kwargs,
        })
        sentry_sdk.capture_exception(exception)

# Route failed tasks to dead-letter queue after max retries
@shared_task(
    bind=True,
    max_retries=3,
    name='payments.charge_card',
)
def charge_card(self, order_id: int) -> None:
    from apps.payments.models import Order, FailedCharge

    try:
        _do_charge(order_id)
    except Exception as exc:
        if self.request.retries >= self.max_retries:
            # Persist to dead-letter table for manual review
            FailedCharge.objects.create(
                order_id=order_id,
                error=str(exc),
                task_id=self.request.id,
            )
            return  # Don't raise — task is permanently failed
        raise self.retry(exc=exc)

Testing Celery Tasks

Unit Testing (No Broker)

# tests/test_tasks.py
import pytest
from unittest.mock import patch, MagicMock
from apps.notifications.tasks import send_welcome_email

class TestSendWelcomeEmail:

    @pytest.mark.django_db
    def test_sends_email_to_existing_user(self, user):
        with patch('apps.notifications.services.EmailService') as mock_email:
            send_welcome_email(user.pk)
            mock_email.send_welcome.assert_called_once_with(user)

    @pytest.mark.django_db
    def test_skips_missing_user_gracefully(self):
        """Should not raise when user is deleted between enqueue and execute."""
        send_welcome_email(99999)  # Non-existent user — must not raise

Integration Testing with CELERY_TASK_ALWAYS_EAGER

# config/settings/test.py
CELERY_TASK_ALWAYS_EAGER = True      # Run tasks synchronously in tests
CELERY_TASK_EAGER_PROPAGATES = True  # Re-raise exceptions from tasks

# tests/test_integration.py
@pytest.mark.django_db
def test_registration_triggers_welcome_email(client):
    with patch('apps.notifications.services.EmailService') as mock_email:
        response = client.post('/api/users/', {
            'email': 'new@example.com',
            'password': 'strongpass123',
        })

    assert response.status_code == 201
    mock_email.send_welcome.assert_called_once()

Testing Retries

@pytest.mark.django_db
def test_task_retries_on_connection_error():
    with patch('apps.crm.services.CRMClient.sync') as mock_sync:
        mock_sync.side_effect = ConnectionError('timeout')

        with pytest.raises(ConnectionError):
            sync_contact_to_crm.apply(args=[1], throw=True)

        assert mock_sync.call_count == 1  # First attempt only when eager

Monitoring

# Inspect active workers and queues
celery -A config inspect active
celery -A config inspect stats
celery -A config inspect reserved

# Check queue lengths (Redis)
redis-cli llen celery

# Flower: web-based real-time monitor
pip install flower
celery -A config flower --port=5555

Anti-Patterns

# BAD: Passing model instances — they may be stale by execution time
send_welcome_email.delay(user)        # Never pass ORM objects
send_welcome_email.delay(user.pk)     # Always pass PKs

# BAD: Calling tasks synchronously in production views
result = generate_report.apply()      # Blocks the request thread

# BAD: Non-idempotent task without guards
@shared_task
def charge_and_fulfill(order_id):
    order.charge()     # May charge twice if task retries!
    order.fulfill()

# GOOD: Idempotent with status guard
@shared_task
def charge_and_fulfill(order_id):
    order = Order.objects.select_for_update().get(pk=order_id)
    if order.status != Order.Status.PENDING:
        return  # Already processed
    order.charge()
    order.fulfill()

Production Checklist

Check
Setting

Worker restarts on crash
supervisord or systemd unit

CELERY_TASK_ACKS_LATE = True
Re-queue tasks on worker crash

CELERY_WORKER_PREFETCH_MULTIPLIER = 1
Fair distribution of long tasks

Separate queues per priority
-Q default,high_priority,low_priority

CELERY_TASK_SOFT_TIME_LIMIT set
Graceful timeout before hard kill

Sentry integration
Capture all task_failure signals

Flower or other monitor
Visibility into queue depths

Beat runs on single node only
Prevents duplicate scheduled task execution

Related Skills

django-patterns — ORM, service layer, and project structure

django-tdd — Testing Django models, views, and services

python-testing — pytest configuration and fixtures

related skills

semantically similar in the cross-vendor index

skills.sh

68% match

django-celery-expert

Expert Django Celery guidance for asynchronous task processing. Use when designing background tasks, configuring Celery workers, handling task retries and…

don't have the plugin yet? install it then click "run inline in claude" again.

django-celery

intent

use this skill when you need background task processing in django. add celery + redis or rabbitmq for offloading slow work (email, pdfs, api calls) from the request cycle, scheduling periodic jobs with beat, retrying failed tasks with exponential backoff, chaining or grouping tasks with canvas, and testing tasks without a broker. production-grade patterns cover idempotency, soft timeouts, dead-letter queues, and monitoring.

inputs

python packages:

celery[redis] or celery[amqp]
django-celery-results
django-celery-beat

environment variables:

CELERY_BROKER_URL: redis://localhost:6379/0 or amqp://guest:guest@localhost//
CELERY_RESULT_BACKEND: django-db, redis, or database-backed

external connections:

redis or rabbitmq broker (network accessible from workers)
sentry or similar (optional, for task failure monitoring)
flower (optional, web ui for real-time monitoring)

django setup:

config/celery.py entrypoint configured with app.autodiscover_tasks()
INSTALLED_APPS includes django_celery_results and django_celery_beat
settings keys: CELERY_BROKER_URL, CELERY_RESULT_BACKEND, CELERY_TASK_TIME_LIMIT, CELERY_TASK_SOFT_TIME_LIMIT, CELERY_TASK_ACKS_LATE, CELERY_BEAT_SCHEDULER

procedure

1. install and initialize celery

inputs: python project with django installed

steps:

run pip install celery[redis] django-celery-results django-celery-beat

create config/celery.py with app initialization:

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.development')
app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

update config/__init__.py to import the celery app:

from .celery import app as celery_app
__all__ = ('celery_app',)

outputs: celery app ready to discover and run tasks

2. configure django settings for celery

inputs: django settings module, redis or rabbitmq broker url

steps:

set CELERY_BROKER_URL to redis or rabbitmq connection string
set CELERY_RESULT_BACKEND to 'django-db' or a separate redis instance
configure serialization: set CELERY_ACCEPT_CONTENT, CELERY_TASK_SERIALIZER, CELERY_RESULT_SERIALIZER to 'json' (safer than pickle)
set timeouts: CELERY_TASK_TIME_LIMIT (hard kill, e.g. 30 min), CELERY_TASK_SOFT_TIME_LIMIT (graceful timeout, e.g. 25 min)
set CELERY_TASK_ACKS_LATE = True to re-queue tasks if a worker crashes
set CELERY_WORKER_PREFETCH_MULTIPLIER = 1 to prevent workers from hoarding long tasks
set CELERY_RESULT_EXPIRES = 86400 (24 hours) to auto-clean task results
add django_celery_results and django_celery_beat to INSTALLED_APPS
set CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler' for db-backed schedules

outputs: django settings configured; celery will respect broker, serialization, and task behavior settings

3. start celery workers

inputs: django project with celery configured

steps for development:

run celery -A config worker --loglevel=info to start a single worker consuming from 'default' queue

steps for production:

run celery -A config worker --loglevel=warning --concurrency=4 -Q default,high_priority to start a worker with multiple concurrency and queue routing
use supervisord, systemd, or container orchestration to restart workers on crash
never run celery worker --beat in production; always run beat on a separate single-node service

outputs: worker process listening on broker, ready to consume and execute tasks

4. start celery beat scheduler

inputs: django project with django_celery_beat installed

steps:

run celery -A config beat --loglevel=info --scheduler django_celery_beat.schedulers:DatabaseScheduler to start beat on a single node
(do not run beat on multiple nodes or use --beat flag in worker; will cause duplicate task executions)

outputs: beat process reading schedules from db, enqueuing periodic tasks at configured times

5. define a basic task

inputs: django app with tasks.py module

steps:

create apps/notifications/tasks.py
import shared_task: from celery import shared_task
decorate a function with @shared_task(name='notifications.send_welcome_email')
pass only primitives and pks, never orm instances
handle missing objects gracefully without raising (idempotent pattern)
log outcomes with logger.info or logger.warning

example:

from celery import shared_task
import logging

logger = logging.getLogger(__name__)

@shared_task(name='notifications.send_welcome_email')
def send_welcome_email(user_id: int) -> None:
    from apps.users.models import User
    from apps.notifications.services import EmailService
    
    try:
        user = User.objects.get(pk=user_id)
    except User.DoesNotExist:
        logger.warning('send_welcome_email: user %s not found', user_id)
        return
    
    EmailService.send_welcome(user)
    logger.info('Welcome email sent to user %s', user_id)

outputs: task function callable via task_name.delay() or .apply_async()

6. define a retryable task with backoff

inputs: task that may fail transiently (api calls, external service syncs)

steps:

use @shared_task(bind=True, max_retries=5, default_retry_delay=60, autoretry_for=(ConnectionError, TimeoutError), retry_backoff=True, retry_backoff_max=600, retry_jitter=True)
catch specific exceptions and call self.retry(exc=exc, countdown=...) to retry with exponential backoff
for rate limits, extract retry_after header from response and pass as countdown

example:

@shared_task(
    bind=True,
    name='integrations.sync_to_crm',
    max_retries=5,
    default_retry_delay=60,
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,
    retry_backoff_max=600,
    retry_jitter=True,
)
def sync_contact_to_crm(self, contact_id: int) -> dict:
    from apps.crm.services import CRMClient
    
    try:
        result = CRMClient().sync(contact_id)
        return result
    except CRMClient.RateLimitError as exc:
        raise self.retry(exc=exc, countdown=int(exc.retry_after))

outputs: task retries up to max_retries with exponential backoff; jitter prevents thundering herd

7. implement idempotent task with status guard

inputs: task that modifies state and may be retried multiple times

steps:

use select_for_update() to acquire a lock on the record
check a status field to ensure the task has not already run (e.g., only update if status == PENDING)
use conditional update: Order.objects.filter(pk=order_id, status=PENDING).update(status=SHIPPED)
if update returns 0 rows, log and return without raising

example:

@shared_task(name='orders.mark_shipped')
def mark_order_shipped(order_id: int, tracking_number: str) -> None:
    from apps.orders.models import Order
    
    updated = Order.objects.filter(
        pk=order_id,
        status=Order.Status.PROCESSING,
    ).update(
        status=Order.Status.SHIPPED,
        tracking_number=tracking_number,
    )
    
    if not updated:
        logger.info('mark_order_shipped: order %s already shipped or not found', order_id)

outputs: task is safe to execute multiple times; no duplicate state changes

8. handle soft timeouts gracefully

inputs: long-running task (pdf generation, report building)

steps:

set both soft_time_limit (e.g. 120 seconds) and time_limit (e.g. 150 seconds) on the task
wrap task body in try/except catching SoftTimeLimitExceeded
in except block, clean up partial files or resources before the hard kill
re-raise the exception to allow celery to mark the task as failed

example:

from celery.exceptions import SoftTimeLimitExceeded

@shared_task(
    bind=True,
    name='reports.generate_pdf',
    soft_time_limit=120,
    time_limit=150,
)
def generate_pdf_report(self, report_id: int) -> str:
    from apps.reports.services import PDFGenerator
    
    try:
        path = PDFGenerator.build(report_id)
        return path
    except SoftTimeLimitExceeded:
        PDFGenerator.cleanup(report_id)
        raise

outputs: task cleans up partial state on timeout; no dangling resources

9. call a task from django code

inputs: task defined, celery worker running

steps:

call task_name.delay(*args, **kwargs) for fire-and-forget async execution
call task_name.apply_async(args=[...], countdown=3600) to schedule in the future
call task_name.apply_async(args=[...], eta=timezone.now() + timedelta(days=1)) to schedule at a specific datetime
call task_name.apply_async(args=[...], queue='high_priority') to route to a specific queue
call task_name.apply(args=[...]) only in tests or debugging (runs synchronously, no broker)

example:

from celery import current_app
from datetime import timedelta
from django.utils import timezone

send_welcome_email.delay(user.pk)
send_reminder.apply_async(args=[user.pk], countdown=3600)
send_reminder.apply_async(args=[user.pk], eta=timezone.now() + timedelta(days=1))
sync_contact_to_crm.apply_async(args=[contact.pk], queue='high_priority')

outputs: task enqueued on broker; worker will execute asynchronously

10. define periodic tasks via code-based beat schedule

inputs: django settings, celery configured

steps:

add CELERY_BEAT_SCHEDULE dict to django settings
for each periodic task, define a dict entry with 'task' (task name), 'schedule' (crontab or int seconds), and optional 'args' or 'kwargs'
use crontab(hour=2, minute=0) for 2am daily, crontab(day_of_week='monday', hour=8, minute=0) for mondays at 8am, or 60.0 for every 60 seconds
beat will read this dict and enqueue tasks at the specified times

example:

from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    'cleanup-expired-sessions': {
        'task': 'users.cleanup_expired_sessions',
        'schedule': crontab(hour=2, minute=0),
    },
    'sync-inventory': {
        'task': 'products.sync_inventory',
        'schedule': 60.0,
    },
    'weekly-digest': {
        'task': 'notifications.send_weekly_digest',
        'schedule': crontab(day_of_week='monday', hour=8, minute=0),
    },
}

outputs: beat enqueues tasks on schedule; no manual triggering needed

11. define periodic tasks via database (dynamic)

inputs: django_celery_beat installed, CELERY_BEAT_SCHEDULER = DatabaseScheduler

steps:

create a CrontabSchedule via orm: CrontabSchedule.objects.get_or_create(hour='*/6', minute='0', timezone='UTC')
create or update a PeriodicTask with the crontab, task name, and args (as json string)
set 'enabled': True to activate; False to pause
beat reads from db periodically (every 5 seconds by default) and enqueues tasks

example:

from django_celery_beat.models import PeriodicTask, CrontabSchedule
import json

schedule, _ = CrontabSchedule.objects.get_or_create(
    hour='*/6', minute='0', timezone='UTC',
)

PeriodicTask.objects.update_or_create(
    name='Sync inventory every 6 hours',
    defaults={
        'crontab': schedule,
        'task': 'products.sync_inventory',
        'args': json.dumps([]),
        'enabled': True,
    }
)

outputs: periodic task registered in db; beat will enqueue at scheduled intervals

12. chain tasks sequentially

inputs: multiple tasks that depend on each other's output

steps:

import chain from celery
build a chain: chain(task1.s(arg1), task2.s(), task3.s())
each task receives the previous task's return value as its first positional argument
call .delay() to enqueue the chain
chain runs sequentially on a single worker; slower than parallel but preserves order and data flow

example:

from celery import chain

pipeline = chain(
    fetch_data.s(source_id),
    transform_data.s(),
    load_to_warehouse.s(),
)
pipeline.delay()

outputs: tasks execute sequentially; result of each task passed to the next

13. group tasks in parallel

inputs: multiple independent tasks

steps:

import group from celery
build a group: group(task1.s(arg1) for arg1 in arg_list)
call .delay() to enqueue all tasks to the same queue
workers execute tasks in parallel
group does not wait or aggregate results

example:

from celery import group

parallel = group(
    send_welcome_email.s(user_id)
    for user_id in new_user_ids
)
parallel.delay()

outputs: all tasks enqueued; workers execute in parallel

14. chord tasks with callback

inputs: multiple tasks that must complete before a final aggregation task

steps:

import chord from celery
build a chord: chord(group(task1.s(arg), ...), callback_task.s())
workers execute all tasks in the group in parallel
when all complete, celery invokes the callback task with a list of results
callback aggregates or processes the results

example:

from celery import chord, group

result = chord(
    group(process_chunk.s(chunk) for chunk in data_chunks),
    aggregate_results.s(),
)
result.delay()

outputs: parallel tasks run; callback invoked with all results once group completes

15. set up sentry error tracking for failed tasks

inputs: sentry project, celery task_failure signal

steps:

install sentry: pip install sentry-sdk
import sentry_sdk and task_failure signal in a task module (e.g., apps/core/tasks.py)
connect a handler to @task_failure.connect that captures the exception with task context
in the handler, create a sentry scope with task name, task_id, args, and kwargs
call sentry_sdk.capture_exception(exception)

example:

from celery.signals import task_failure
import sentry_sdk

@task_failure.connect
def on_task_failure(sender, task_id, exception, args, kwargs, traceback, einfo, **kw):
    with sentry_sdk.new_scope() as scope:
        scope.set_context('celery', {
            'task': sender.name,
            'task_id': task_id,
            'args': args,
            'kwargs': kwargs,
        })
        sentry_sdk.capture_exception(exception)

outputs: all task failures logged to sentry with full context

16. route failed tasks to dead-letter queue

inputs: task that may fail after max retries; FailedTask or similar model for persistence

steps:

define a task with max_retries
on final exception (after retries exhausted), do not raise
instead, persist the failure to a dead-letter table (FailedCharge, FailedPayment, etc.) with order_id, error message, and task_id
return gracefully so the task is not retried further
separately, query the dead-letter table for manual review or async retry

example:

@shared_task(bind=True, max_retries=3, name='payments.charge_card')
def charge_card(self, order_id: int) -> None:
    from apps.payments.models import Order, FailedCharge
    
    try:
        _do_charge(order_