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.

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_COMPRESS_THRESHOLD	1024	Bytes threshold for Zstandard compression
VTASKS_DLQ_CAP	1000	Maximum failed tasks in Dead Letter Queue
VTASKS_VALKEY_PREFIX	"vt"	Prefix for Valkey keys (namespace isolation)
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)
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 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)
    },
}