Backend

Background tasks

Celery runs emails, webhook deliveries and housekeeping. One worker with beat is all a project needs.

Work that should not slow down a request runs on Celery: sending emails, delivering webhooks, periodic housekeeping. The base settings configure Celery for you (CELERY_* keys), and pleate_backend.settings.redis(url) points the broker at Redis.

Running a worker

myproject/celery.py

import os
from celery import Celery

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")

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

# development: one process with the beat scheduler embedded
celery -A myproject worker -B -l info

# production: workers and exactly one beat
celery -A myproject worker -l info -c 4
celery -A myproject beat -l info

Tasks are acknowledged late and requeued if a worker dies (CELERY_TASK_ACKS_LATE, CELERY_TASK_REJECT_ON_WORKER_LOST), so write them to be safe to run twice. Results are not stored.

Built in tasks

Task	When	Does
pleate_backend.emails.send	On every send_email	Renders and sends one email. Retries five times with backoff on network errors.
pleate_backend.webhooks.tasks.deliver	On every dispatch	Posts one webhook delivery.
pleate_backend.webhooks.tasks.retry_due	Every minute (beat)	Retries failed deliveries whose next attempt is due.
pleate_backend.core.tasks.cleanup	Every hour (beat)	Runs every registered cleanup function.

pleate_backend/settings.py

CELERY_BEAT_SCHEDULE = {
    "pleate-cleanup": {"task": "pleate_backend.core.tasks.cleanup", "schedule": 60 * 60},
    "pleate-webhook-retries": {"task": "pleate_backend.webhooks.tasks.retry_due", "schedule": 60},
}

Add your own entries by extending the dict in your settings:

python

from celery.schedules import crontab

CELERY_BEAT_SCHEDULE |= {
    "weekly-report": {"task": "reports.tasks.weekly", "schedule": crontab(hour=7, minute=0, day_of_week=1)},
}

The cleanup registry

Housekeeping that belongs to an app registers itself instead of adding a beat entry. The files app uses it to remove direct uploads that were never completed:

myapp/handlers.py

from datetime import timedelta
from django.utils import timezone
from pleate_backend.core.tasks import register_cleanup

@register_cleanup
def drop_abandoned_carts():
    Cart.objects.filter(updated_at__lt=timezone.now() - timedelta(days=30)).delete()

Each function runs in its own try: one failing cleanup is logged and the others still run.

Writing tasks

Name tasks explicitly, @shared_task(name="myapp.tasks.export"), so renaming a module does not orphan queued messages.
Pass ids, not model instances, and reload inside the task.
Make them idempotent: with late acknowledgement a task may run again after a crash.
Tell the user when a long task ends with notify(); it reaches the open page through SSE.

python

from celery import shared_task
from pleate_backend.notifications.services import notify

@shared_task(name="reports.tasks.export_orders")
def export_orders(user_id):
    user = User.objects.get(pk=user_id)
    file = build_export(user)                     # e.g. save_upload(...) into the files app
    notify(user, "export.ready", "Your export is ready", url=f"/files/{file.pk}")

Without a worker

In tests set CELERY_TASK_ALWAYS_EAGER = True and tasks run inline. Without Redis in development, set PLEATE["EMAILS"]["ASYNC"] = False: emails are sent inline, and webhook deliveries are still recorded and retried once a worker runs.