DOCSEARCHbyAlgolia
Submit Docs

django-db-queue

repository·master·Indexed 41 days ago

https://github.com/dabapps/django-db-queue

A simple, database-backed job queue for Django that supports multi-step job workflows and asynchronous tasks. It provides a system for defining jobs and tasks, managing job states (NEW, READY, PROCESSING, STOPPING, COMPLETED, FAILED), and utilizing workspaces for inter-task communication. Compatible with Django 4.2, 5.2, 6.0 and Python 3.10 through 3.14.

Tokens
3.4K
Snippets
15
Records
19
Agent score
48%

What's inside django-db-queue

  1. Use failure, creation, and task hooks

    master

    Hooks allow you to execute code at different stages of a job's lifecycle. They are configured in the JOBS dictionary in your settings.

    Hook Types

    Hook TypeConfig KeyExecution ContextArgumentsBehavior
    Failurefailure_hookWorker process(job, exception)Runs when an unhandled exception occurs.
    Creationcreation_hookProcess that created the job(job)Runs immediately after Job.objects.create() call.
    Pre-Taskpre_task_hookWorker process(job)Runs before the task. If it fails, the job is marked FAILED.
    Post-Taskpost_task_hookWorker process(job)Runs after the task. Always runs, even on failure. If it fails, the job is not marked as failed.

    Implementation Example

    # settings.py
JOBS = {
    "my_job": {
        "tasks": ["project.common.jobs.my_task"],
        "failure_hook": "project.common.jobs.my_task_failure_hook",
        "creation_hook": "project.common.jobs.my_task_creation_hook",
        "pre_task_hook": "project.common.jobs.my_pre_task_hook",
        "post_task_hook": "project.common.jobs.my_post_task_hook",
    },
}

  2. Core Concepts: Jobs, Tasks, and Workspaces

    master

    Understanding the relationship between the three primary abstractions in django-db-queue is essential for implementing background work:

    1. Job: The top-level unit of work, represented as a Django model instance stored in the database.
    2. Task: A Python function that processes a Job. Tasks must accept exactly one argument: the Job instance. A single Job can consist of a sequence of multiple tasks.
    3. Workspace: A Python dictionary attached to the Job instance (job.workspace). It serves two purposes:
      • Argument Passing: Providing metadata or parameters (like API tokens or IDs) to tasks.
      • Inter-task Communication: Tasks can modify the workspace, and these changes are passed to subsequent tasks in the job sequence.

    Workspaces are stored as JSON and can be queried using standard Django __ lookups.

    # Creating a job with a workspace
Job.objects.create(name="my_job", workspace={"user_id": 123})

# Accessing workspace in a task
def my_task(job):
    user_id = job.workspace["user_id"]
    # ... do work ...

# Querying jobs by workspace content
user_jobs = Job.objects.filter(workspace__user_id=123)

  3. Define and configure a job

    master

    Jobs in django-db-queue are defined in two steps: writing the task function and registering it in your Django settings.

    1. Define the task: Create a Python function that accepts a job instance as its argument.
    2. Register the job: In your settings.py, add a JOBS dictionary where the key is your job name and the value is a configuration dictionary containing a tasks list of import paths to your task functions.
    # project/common/jobs.py
def my_task(job):
    # task logic here
    pass

# settings.py
JOBS = {
    "my_job": {
        "tasks": ["project.common.jobs.my_task"],
    },
}

  4. Install and set up django-db-queue

    master

    To use django-db-queue, install it via pip, add django_dbq to your INSTALLED_APPS, and run your Django migrations.

    Compatibility:

    • Django 4.2, 5.2, 6.0
    • Python 3.10 through 3.14

    Note: If you require Django 2.2 support, use the 1.x release series.

    pip install django-db-queue
    # settings.py
INSTALLED_APPS = [
    ...,
    "django_dbq",
]
    python manage.py migrate

  5. Configure Job Task Sequences

    master

    To define how a job is processed, you must map a job name to a list of task function paths in your Django settings.py. The worker will execute these tasks in the order provided.

    Example configuration mapping the job name import_cats to two sequential steps:

    JOBS = {
    'import_cats': [
        'apps.cat_importer.import_cats.step_one',
        'apps.cat_importer.import_cats.step_two'
    ],
}

  7. Avoid `bulk_create` with Job models

    master

    The Job model contains logic in its save() method that is bypassed when using Django's bulk_create(). Using bulk_create will result in jobs that lack the necessary next_task information to progress.

    If you must use bulk_create for performance, you must manually compute the next_task for each instance using django_dbq.tasks.get_next_task_name based on the job's name. Note that creation_hook will not be triggered when using this method.

  8. Create, prioritise, and schedule jobs

    master

    You interact with the queue by creating Job instances via the Django ORM using the name configured in your JOBS settings.

    Basic Creation

    Job.objects.create(name="my_job")

    Prioritising Jobs

    Set the priority field to an integer higher than 0 (default is 0). Jobs are processed from highest priority to lowest, then by creation time (oldest first).

    Job.objects.create(name="important_job", priority=1)

    Scheduling Jobs

    Use the run_after field to specify a future datetime when the job should run. The worker will pick up the job once the time has passed.

    from django.utils import timezone
from datetime import timedelta

Job.objects.create(
    name="scheduled_job",
    run_after=(timezone.now() + timedelta(minutes=10)),
)
    # Create a high priority job
Job.objects.create(name="critical_job", priority=2)

# Schedule a job for 10 minutes from now
Job.objects.create(
    name="scheduled_job",
    run_after=(timezone.now() + timedelta(minutes=10)),
)

  9. Monitor Queue Depths

    master

    You can monitor the number of pending jobs in your queues using either the Python API or the CLI.

    Python API

    Use Job.get_queue_depths() to get a dictionary mapping queue names to the number of jobs in NEW or READY states.

    Note: If a queue has zero jobs, it will not be included in the returned dictionary. Use exclude_future_jobs=True to ignore jobs scheduled for the future.

    CLI

    Use the queue_depth command to check depths from the terminal. You can pass one or more queue names as arguments.

    # Python API
from django_dbq.models import Job

# Get depths for all queues
depths = Job.get_queue_depths()

# Get depths excluding future scheduled jobs
depths = Job.get_queue_depths(exclude_future_jobs=True)
    # CLI: Check depth for specific queues
python manage.py queue_depth default my_other_queue

# CLI: Exclude future scheduled jobs
python manage.py queue_depth default --exclude_future_jobs

  10. Job States Reference

    master

    Jobs transition through several states during their lifecycle:

    • NEW: Created and waiting for a worker to pick up the first task.
    • READY: A task has completed, and the job is awaiting the next task in the sequence.
    • PROCESSING: A worker is currently executing a task.
    • STOPPING: The worker has received an OS signal to exit.
    • COMPLETED: All tasks in the job sequence finished successfully.
    • FAILED: A task within the job failed.

  12. Clean up Old Jobs

    master

    To prevent the Job table from growing indefinitely, use the delete_old_jobs management command. By default, it deletes jobs in COMPLETED or FAILED states that are older than 24 hours. You can adjust this threshold using the --hours flag.

    # Delete jobs older than 24 hours (default)
python manage.py delete_old_jobs

# Delete jobs older than 12 hours
python manage.py delete_old_jobs --hours 12