Configuration

Backend Configuration

Configure your task backend in Django settings using the TASKS dictionary.

Database Backend

INSTALLED_APPS = [
    # ...
    "django_vtasks",
    "django_vtasks.db",  # Required for Database backend
]

TASKS = {
    "default": {
        "BACKEND": "django_vtasks.backends.db.DatabaseTaskBackend",
    }
}

Database Compatibility

The Database backend relies on SELECT ... FOR UPDATE SKIP LOCKED for efficient parallel processing. SQLite and older MySQL versions don't support this, limiting them to one worker at a time.

Connection lifetime

The worker's polling loops hold a database connection of their own, separate from the ones task bodies use. It is recycled when a poll fails, so a connection dropped by a pooler, a failover or an idle timeout costs one failed fetch rather than disabling the worker, and again on a ~60s cadence so it does not live forever.

What that recycle costs depends on how the database is configured, and both supported shapes are cheap:

• With a connection pool (OPTIONS={"pool": True}) the recycle hands the connection back and the next poll borrows one. No handshake, and a mostly-idle worker stops holding a slot it is not using.
• Without a pool — Django's default — the recycle closes the connection and the next poll reconnects: one connect per worker process per minute.

A pool is worth considering for an ASGI deployment generally, not just for the worker: ASGIHandler runs each request in its own ThreadSensitiveContext, so requests do not share a connection and each one otherwise pays for its own.

Valkey Backend

INSTALLED_APPS = [
    # ...
    "django_vtasks",
]

TASKS = {
    "default": {
        "BACKEND": "django_vtasks.backends.valkey.ValkeyTaskBackend",
        "OPTIONS": {
            "BROKER_URL": "valkey://localhost:6379/0",
            # Optional: Timeout for blocking operations (default: 1.0)
            "BLOCKING_TIMEOUT": 1.0,
        }
    }
}

BLOCKING_TIMEOUT is the maximum wait time when queues are idle. Tasks that arrive are processed immediately regardless of this value.

• Production: 1.0 second (default) - only 1 Redis request per second per worker when idle
• Testing: Use 0.1 seconds for faster test execution

Shared Cache Connection

If you use a compatible cache backend like django-vcache, share connections to minimize resource usage:

CACHES = {
    "default": {
        "BACKEND": "django_vcache.backend.ValkeyCache",
        "LOCATION": "valkey://localhost:6379/1",
    },
}

TASKS = {
    "default": {
        "BACKEND": "django_vtasks.backends.valkey.ValkeyTaskBackend",
        "OPTIONS": {
            "cache_alias": "default",
        }
    }
}

Shared Connection Pool

For applications using valkey-py directly, share an existing connection pool:

import valkey.asyncio as valkey

MY_APP_VALKEY_POOL = valkey.ConnectionPool.from_url("valkey://localhost:6379/0")

TASKS = {
    "default": {
        "BACKEND": "django_vtasks.backends.valkey.ValkeyTaskBackend",
        "OPTIONS": {
            "CONNECTION_POOL": MY_APP_VALKEY_POOL,
            # Still required for synchronous operations like task.enqueue()
            "BROKER_URL": "valkey://localhost:6379/0",
        }
    }
}

Settings Reference

Setting	Default	Description
VTASKS_QUEUES	["default"]	Queues the worker processes — a list of names, or a dict of name → options (worker_concurrency, batch). See Queue configuration
VTASKS_CONCURRENCY	20	Global per-worker concurrency: default per-queue limit and the shared pool for queues without their own worker_concurrency
VTASKS_RUN_SCHEDULER	True	Whether to run the scheduler when requested
VTASKS_SCHEDULE	{}	Periodic task schedules
VTASKS_DLQ_CAP	1000	Maximum failed tasks in Dead Letter Queue
VTASKS_MAX_RESCUES	0	How many times a task may be reclaimed after a worker death before going to the dead-letter queue. At the default of 0 a crashed task is dead-lettered, not retried — raise it only for tasks that are safe to run twice, since a reclaimed task may have partially completed
VTASKS_RESCUE_INTERVAL	30	Seconds between periodic sweeps for tasks orphaned by dead siblings (one worker per deployment sweeps, via a distributed lock). 0 disables the sweep, leaving only the startup scan
VTASKS_HEARTBEAT_INTERVAL	5	Seconds between worker liveness heartbeats; the marker's TTL is 3x this. Rescue uses it to tell a dead sibling from a live one. 0 disables heartbeats, which makes any different-PID sibling look dead — unsafe with more than one worker per host
VTASKS_VALKEY_PREFIX	"{vt}"	Prefix for Valkey keys (namespace isolation). Braces are a Valkey Cluster hash tag — keep them to keep a deployment's keys in one slot
VTASKS_METRICS_PORT	None	Port for Prometheus metrics (standalone workers)
VTASKS_HEALTH_CHECK_FILE	None	Path to file touched for liveness probes
VTASKS_WORKER_ID	None	Custom worker ID (defaults to {hostname}-{pid}). Must be unique per live worker — see One worker per ID
VTASKS_BACKEND	"default"	The alias in TASKS to use for the worker
VTASKS_TASK_CONTEXT	None	Per-task context hook — a callable (or dotted path) returning a context manager entered around each task execution. See Per-task context hook

Per-task context hook

VTASKS_TASK_CONTEXT wraps every task execution in a context manager of your choosing — for scoping error-tracker state, log context, or a tracing span to one task. Set it to a callable, or a dotted path to one:

VTASKS_TASK_CONTEXT = "myapp.observability.task_scope"

The worker calls it once per task, and once per function-group in a batch queue, then enters the context manager it returns:

factory(task_id=..., task_ids=..., name=..., queue=...)

Argument	Value
task_id	The task's ID; None for a batch group
task_ids	List of IDs in a batch group; None for a single task
name	Dotted path of the task function
queue	Queue the task was fetched from

Accept **kwargs so later additions don't break your factory.

# myapp/observability.py
from contextlib import contextmanager
import sentry_sdk

@contextmanager
def task_scope(*, name, queue, task_id=None, task_ids=None, **_):
    with sentry_sdk.isolation_scope() as scope:
        scope.clear()  # drop breadcrumbs inherited from the ambient scope
        scope.set_tag("vtasks.task", name)
        scope.set_tag("vtasks.queue", queue)
        yield

The context encloses the task body, the task_started / task_finished / task_failure signals, and the worker's own failure logging — so everything recorded about the task is scoped to it. It cannot suppress task exceptions: __exit__'s return value is ignored. A factory that raises is logged, and the task runs without the context rather than failing.

Two constraints follow from where the hook sits:

• __enter__ and __exit__ run on the worker's event loop, so they must not block. Keep the factory to in-memory bookkeeping — contextvars, scope objects — not I/O. Anything that needs to await belongs in a receiver for the django_vtasks.signals task signals, which vtasks dispatches with asend.
• It wraps worker execution, so it never fires under ImmediateBackend, which runs tasks inline at enqueue time. Tests using immediate mode won't exercise it.

Signals and this hook are complementary: signals are point-in-time notifications at task boundaries, while the hook is a paired enclosure around the whole execution — which is what scoped state (contextvars, error-tracker scopes) requires and a fire-and-forget receiver cannot provide.

Worker Command Arguments

Most arguments can also be set via environment variables:

Argument	Environment Variable	Django Setting
--concurrency	VTASKS_CONCURRENCY	VTASKS_CONCURRENCY
--backend	VTASKS_BACKEND	VTASKS_BACKEND
--id	VTASKS_WORKER_ID	VTASKS_WORKER_ID
--health-check-file	VTASKS_HEALTH_CHECK_FILE	VTASKS_HEALTH_CHECK_FILE
--metrics-port	VTASKS_METRICS_PORT	VTASKS_METRICS_PORT

Queue Configuration

VTASKS_QUEUES declares which queues a worker consumes. It accepts either a list of names or a dict mapping each name to a per-queue options dict:

All concurrency in vtasks is per-worker (per-process). A limit of N means up to N at once in each worker; the cluster-wide ceiling is N times your worker (pod) count.

VTASKS_CONCURRENCY = 50              # global pool / default per-queue limit, per worker

VTASKS_QUEUES = {
    "default": {},                              # shares the global pool
    "cold_storage": {"worker_concurrency": 3},  # its own cap: 3 at once per worker
    "emails": {"batch": {"count": 100, "timeout": 5.0}},
}

# the simple list form still works (all queues share the global pool):
# VTASKS_QUEUES = ["default", "cold_storage"]

Per-queue options (unknown keys raise ImproperlyConfigured so typos can't silently drop a cap):

• worker_concurrency (int) — give this queue its own dedicated semaphore of that size, per worker. Queues without it share the global VTASKS_CONCURRENCY pool.
• batch ({"count", "timeout"}) — process this queue in batches: collect up to count tasks, waiting at most timeout seconds, then hand them to the task as a list.

Per-queue concurrency

VTASKS_CONCURRENCY alone is a single global pool shared by every queue a worker consumes — ideal for cheap I/O-bound tasks. But a handful of heavy CPU/RAM-bound tasks (analytics, image processing, data exports) at that same concurrency can exhaust memory or a connection pool. A queue's own worker_concurrency isolates it:

• A queue with worker_concurrency gets its own semaphore; queues without one keep sharing the global pool, so a saturated capped queue cannot starve the rest.
• A worker's maximum concurrency is VTASKS_CONCURRENCY plus the sum of the per-queue overrides it consumes; the connection-isolation lane pool is sized to match.
• Every limit is per-worker (per-process) — the right scope for bounding per-pod resources like memory or database connections. The fleet-wide ceiling is the limit times your worker count. The worker_ prefix in the key name is a deliberate reminder of this scope at the point where you set it.

Batch processing

Declare batch per queue (see example above). Tasks on a batch queue are collected and delivered as a list — see the Guide.

Periodic Task Configuration

Define scheduled tasks in VTASKS_SCHEDULE:

from datetime import timedelta

from django_vtasks.scheduler import crontab

VTASKS_SCHEDULE = {
    "daily_report": {
        "task": "myapp.tasks.generate_report",
        "schedule": crontab(hour=5, minute=0),
    },
    "hourly_cleanup": {
        "task": "myapp.tasks.cleanup",
        "schedule": 3600,  # Every hour (in seconds)
    },
    "four_hourly_sweep": {
        "task": "myapp.tasks.sweep",
        "schedule": timedelta(hours=4),  # Also accepted
    },
}

Only task and schedule are used. Celery beat's args, kwargs and options have no equivalent — the scheduler always enqueues with no arguments.

A schedule is a cron string (crontab() builds one), a number of seconds, or a timedelta. Anything else the scheduler cannot act on. Rather than leave the entry quietly never due, it logs an error naming the task and moves on, and repeats that at most hourly for as long as the entry exists. The same applies to a task path that cannot be resolved.

Validating the schedule

manage.py check validates the whole of VTASKS_SCHEDULE, so a mistake surfaces in development or CI instead of as a task that never runs:

ID	Reported when
django_vtasks.W001	VTASKS_SCHEDULE is not a mapping
django_vtasks.W002	An entry is not a mapping
django_vtasks.W003	task is not a dotted path to a function
django_vtasks.W004	An entry has no schedule
django_vtasks.W005	A cron string does not resolve to a run time (malformed, or valid but never occurring)
django_vtasks.W006	A schedule is a type the scheduler cannot act on, or a number that is not finite
django_vtasks.W007	A schedule is a bool — an int in Python, so a one-second interval
django_vtasks.W008	An interval is zero or less, so the task is due on every tick
django_vtasks.W009	An entry has keys beyond task and schedule, which are ignored

These are warnings, never errors. An error-level check aborts migrate and runserver, which would fail the upgrade of exactly the deployments this validation is for — the ones whose schedule holds something an earlier release ignored without saying so. To make them a gate in CI:

python manage.py check --fail-level WARNING

Run only these with python manage.py check --tag vtasks, and silence one you disagree with through Django's SILENCED_SYSTEM_CHECKS:

SILENCED_SYSTEM_CHECKS = ["django_vtasks.W008"]

None of this is imported at startup, let alone run: the app registers a stub and the check module loads only in a process that actually runs a check, which is a management command. Neither the ASGI app nor runworker runs system checks, so a deployment pays nothing for any of it on boot.