django-task-psql

A Postgres-native backend for Django Tasks (django.tasks, Django 6.0+).

Django 6.0 introduced an official interface for background task queues (@task, .enqueue(), pluggable backends via the TASKS setting). django-task-psql is a production backend for that interface, built specifically for Postgres: real queueing via SKIP LOCKED, LISTEN/NOTIFY wakeup instead of polling, and automatic retries with exponential backoff — no Celery or Redis required.

Built Postgres-only on purpose, trading multi-engine support for tighter integration with Postgres features (see django-tasks-db for a database-agnostic alternative).

Installation

pip install django-task-psql

INSTALLED_APPS = [
    ...,
    "django_task_psql",
]

TASKS = {
    "default": {
        "BACKEND": "django_task_psql.backend.PostgresBackend",
        "QUEUES": ["default"],
        "OPTIONS": {
            # Optional. Defaults to uuid.uuid4.
            "id_function": "uuid.uuid4",
            # Optional. Default attempts before a task is marked FAILED. Defaults to 1.
            "max_attempts": 3,
        },
    }
}

python manage.py migrate

Usage

Standard django.tasks API — no custom decorator:

from django.tasks import task

@task
def send_welcome_email(user_id):
    ...

send_welcome_email.enqueue(user_id)

Per-task retry override: @task(queue_name="emails", max_attempts=1). Overrides OPTIONS["max_attempts"] for that task only; tasks that don't set it keep using the backend-wide default.

Coroutines (async def) work out of the box — Task.call() bridges them via async_to_sync internally, no asyncio event loop needed in the worker.

Running the worker

python manage.py runworker --queues default emails --concurrency 4

• --batch: drain the queue then exit (useful for a Kubernetes Job).
• --max-tasks N: exit after roughly N tasks.
• --backend default: which TASKS alias to process.

Configuration via environment variables

All of these have sane defaults for local development — set them in production as needed:

A CLI flag (--concurrency, --queues) always overrides its corresponding environment variable.

Dead-letter queue

python manage.py dlq_list                    # all failed tasks
python manage.py dlq_list --queue emails --limit 50
python manage.py dlq_list --json

python manage.py dlq_replay <id>             # requeue one failed task
python manage.py dlq_replay --all --queue emails

python manage.py stats --days 7              # queue stats: totals + top failing tasks
python manage.py stats --queue emails --json

python manage.py cleanup_tasks --days 7      # prune old finished rows, e.g. from cron

Django admin

TaskRow is registered read-only in the Django admin for inspection.

OpenTelemetry (optional)

pip install django-task-psql[otel]

Installing the otel extra (opentelemetry-api only) wraps every task execution in a span named <TASK_WORKER_SPAN_PREFIX>.<task_path> (attributes: django_task_psql.attempt, django_task_psql.queue_name), with exceptions recorded and the span marked as an error. Nothing is exported unless your app configures a TracerProvider and exporter itself (e.g. opentelemetry-sdk + an OTLP exporter); without opentelemetry-api installed at all, tracing is skipped entirely with zero import errors.

How it works

• Claiming: UPDATE ... WHERE id = (SELECT ... FOR UPDATE SKIP LOCKED LIMIT 1) RETURNING ... — safe for multiple worker processes running in parallel.
• Wakeup: a Postgres trigger emits NOTIFY task_psql_new on every INSERT of a READY task (fired after commit, so tasks created inside a transaction only wake the worker once it commits). The worker blocks on LISTEN with a heartbeat fallback (TASK_WORKER_HEARTBEAT_S) to catch deferred tasks whose run_after has elapsed without a fresh NOTIFY.
• Concurrency: the worker's main thread claims one task at a time and submits it to a ThreadPoolExecutor; a Semaphore limits how many are in flight before claiming the next. Each thread closes its own database connection when it's done with a task — with Django's native connection pool enabled, that returns the connection to the pool instead of dropping the socket, which avoids a connection leak that recurring background threads are prone to.
• Retries: on failure, a task is rescheduled with run_after = now() + backoff_base * 2^(attempt - 1), until max_attempts is reached, at which point it's marked FAILED. max_attempts can be set per-task (@task(max_attempts=1)), falling back to OPTIONS["max_attempts"] when unset.

Scope

Targets Postgres and Django 6.0+. Designed for a single Postgres cluster; SKIP LOCKED already supports running multiple worker processes against it in parallel.

License

MIT