celery-patterns

---

name: celery-patterns description: Enforces Celery task patterns: idempotent design, retry strategies, error handling, testing with eager mode. Activates when working with Celery tasks, background jobs, or async workers. allowed-tools: Read, Grep, Glob, Bash(ruff:), Bash(mypy:), Bash(basedpyright:), Bash(make:), Bash(uv:), Bash(python:), Bash(pytest:), Bash(celery:)

Celery Patterns

These rules are NON-NEGOTIABLE. Violating any of them is a bug.

Task Design:

• ALWAYS make tasks idempotent — calling N times produces the same result
• NEVER pass Django model instances as task arguments — ALWAYS pass IDs (primitive types: int, str, UUID)
• NEVER create long tasks without decomposition — ALWAYS split into subtasks for collection operations
• ALWAYS use bind=True for tasks needing self (retry, task info)
• ALWAYS use explicit name parameter — NEVER auto-generated task names (break on refactor)
• ALWAYS use acks_late=True + reject_on_worker_lost=True for tasks that must execute (at-least-once delivery)

Retry & Error Handling:

• NEVER use bare except in tasks — catch specific exceptions
• ALWAYS use autoretry_for + retry_backoff=True + retry_backoff_max + max_retries for retryable errors
• ALWAYS use retry_jitter=True to avoid thundering herd
• NEVER retry business logic errors (validation, not found) — only transient errors (network, external API)
• ALWAYS implement dead letter handling — on_failure callback or monitoring for failed tasks

Organization:

• ALWAYS use shared_task (not app.task) for reusability
• ALWAYS put tasks in <app>/tasks.py — one file per Django app
• NEVER create circular imports — task imports service/model, not the other way

Configuration:

• ALWAYS use task_always_eager=True in test settings (synchronous execution)
• ALWAYS use task_eager_propagates=True in tests (propagate exceptions)
• ALWAYS use CELERY_ namespace prefix in Django settings (config_from_object('django.conf:settings', namespace='CELERY'))
• For non-Django projects (e.g., FastAPI): use celery_app.conf.update() with explicit config dict

Quality:

• ALWAYS run the project's typecheck and test commands after any task change

Celery patterns for background task processing. Framework-independent — works with Django, FastAPI, or standalone Python. All patterns target Python 3.13+, Celery 5.4+.

Task Structure

One task per operation. Use shared_task with explicit name, bind=True, and retry configuration.

# apps/orders/tasks.py
import logging
from uuid import UUID

import httpx
from celery import shared_task
from celery.utils.log import get_task_logger

from apps.notifications.exceptions import NotificationDeliveryError
from apps.orders.models import Order
from apps.orders.services import OrderService

logger: logging.Logger = get_task_logger(__name__)


@shared_task(
    bind=True,
    name="orders.send_order_confirmation",
    acks_late=True,
    reject_on_worker_lost=True,
    autoretry_for=(NotificationDeliveryError, httpx.TransportError),
    retry_backoff=True,
    retry_backoff_max=600,
    retry_jitter=True,
    max_retries=5,
)
def send_order_confirmation(self, order_id: int) -> None:
    """Send confirmation notification for a placed order."""
    try:
        order = Order.objects.get(pk=order_id)
    except Order.DoesNotExist:
        # Not found = business logic error; do NOT retry
        logger.error("Order %s not found, skipping confirmation", order_id)
        return

    OrderService.send_confirmation(order=order)
    logger.info("Confirmation sent for order %s", order_id)

Call by passing the ID, never the model instance:

# correct
send_order_confirmation.delay(order.pk)

# wrong — never pass model instances
send_order_confirmation.delay(order)

Idempotency Patterns

Tasks may execute more than once (retries, worker restarts). Design for safety.

Check-before-act — guard with a status check:

@shared_task(bind=True, name="orders.charge_order", acks_late=True, reject_on_worker_lost=True)
def charge_order(self, order_id: int) -> None:
    order = Order.objects.get(pk=order_id)
    if order.status != Order.Status.PENDING:
        logger.info("Order %s already processed (status=%s), skipping", order_id, order.status)
        return
    OrderService.charge(order=order)

DB uniqueness constraint — let the database enforce deduplication:

# Use update_or_create / get_or_create with a unique key derived from the operation
Payment.objects.get_or_create(
    order_id=order_id,
    defaults={"amount": amount, "status": Payment.Status.PENDING},
)

Idempotency matters because Celery guarantees at-least-once delivery with acks_late=True. A task that charges a card or sends an email must be safe to call twice.

Task Decomposition

Long-running tasks over collections must be split into per-item subtasks. Use group() for fan-out.

# apps/newsletters/tasks.py
from celery import group, shared_task

from apps.newsletters.models import Subscriber
from apps.newsletters.services import NewsletterService


@shared_task(
    name="newsletters.send_newsletter_batch",
    acks_late=True,
    reject_on_worker_lost=True,
)
def send_newsletter_batch(newsletter_id: int) -> None:
    """Fan-out: spawn one task per subscriber."""
    subscriber_ids = list(
        Subscriber.objects.filter(active=True).values_list("id", flat=True)
    )
    job = group(send_newsletter_to_subscriber.s(newsletter_id, sid) for sid in subscriber_ids)
    job.apply_async()


@shared_task(
    bind=True,
    name="newsletters.send_newsletter_to_subscriber",
    acks_late=True,
    reject_on_worker_lost=True,
    autoretry_for=(smtplib.SMTPException, ConnectionError),
    retry_backoff=True,
    retry_backoff_max=300,
    retry_jitter=True,
    max_retries=3,
)
def send_newsletter_to_subscriber(self, newsletter_id: int, subscriber_id: int) -> None:
    NewsletterService.deliver(newsletter_id=newsletter_id, subscriber_id=subscriber_id)

group() schedules all subtasks in parallel. Each item task handles its own retries independently.

Configuration

Django settings

Use the CELERY_ namespace prefix so config_from_object picks up all settings automatically.

# config/settings/base.py
CELERY_BROKER_URL = env("CELERY_BROKER_URL", default="redis://localhost:6379/0")
CELERY_RESULT_BACKEND = env("CELERY_RESULT_BACKEND", default="redis://localhost:6379/1")
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
CELERY_ACCEPT_CONTENT = ["json"]
CELERY_TIMEZONE = "UTC"
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = 300  # hard limit in seconds
CELERY_TASK_SOFT_TIME_LIMIT = 240  # soft limit — raises SoftTimeLimitExceeded

# config/settings/test.py
from config.settings.base import *  # noqa: F403

CELERY_TASK_ALWAYS_EAGER = True
CELERY_TASK_EAGER_PROPAGATES = True

Celery app setup (Django)

# config/celery.py
import os

from celery import Celery

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.local")

app = Celery("myproject")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

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

__all__ = ["celery_app"]

Non-Django projects (FastAPI, standalone)

# celery_app.py
from celery import Celery

app = Celery("myproject")
app.conf.update(
    broker_url="redis://localhost:6379/0",
    result_backend="redis://localhost:6379/1",
    task_serializer="json",
    accept_content=["json"],
    timezone="UTC",
)

Testing Celery Tasks

With task_always_eager=True, task.delay() executes synchronously in the same process — no broker needed.

Preferred approach: test service logic directly, not through the task. The task is a thin wrapper.

# apps/orders/tests/test_services.py
def test_send_confirmation_delivers_email(order: Order) -> None:
    # Test the service directly — fast, no Celery involved
    OrderService.send_confirmation(order=order)
    # assert side effects ...

Task-level test: verify the task calls the service correctly.

# apps/orders/tests/test_tasks.py
import pytest
from unittest.mock import patch

from apps.orders.tasks import send_order_confirmation
from apps.orders.tests.factories import OrderFactory


@pytest.mark.django_db
def test_send_order_confirmation_calls_service() -> None:
    order = OrderFactory()

    with patch("apps.orders.tasks.OrderService.send_confirmation") as mock_send:
        send_order_confirmation.delay(order.pk)  # eager — runs synchronously
        mock_send.assert_called_once_with(order=order)


@pytest.mark.django_db
def test_send_order_confirmation_skips_missing_order() -> None:
    # Should not raise — missing orders are silently skipped
    send_order_confirmation.delay(99999)

Use @override_settings for one-off eager overrides in Django test suites (Django-specific; not applicable to non-Django projects):

from django.test import override_settings

@override_settings(CELERY_TASK_ALWAYS_EAGER=True, CELERY_TASK_EAGER_PROPAGATES=True)
def test_task_runs_eagerly() -> None:
    ...