Task Model

The Task model represents a unit of work to be executed asynchronously by a worker process.

API Reference

Task

Bases: Model

Represents a task to be executed asynchronously by the worker.

A Task stores all the information needed to execute a callable function with the specified arguments, along with its execution state and results.

Attributes:

Name	Type	Description
id		UUID primary key for the task.
created		Timestamp when the task was created.
modified		Timestamp when the task was last modified.
task		Dotted path to the callable (e.g., "myapp.tasks.send_email").
args		JSON-serialized keyword arguments for the callable.
status		Current execution status (QUEUED, PROGRESS, COMPLETED, FAILED, CANCELLED).
output		Return value from the callable (stored as text).
worker_pid		Process ID of the worker handling this task.
error		Error message and traceback if the task failed.
log		Captured stdout/stderr/logging output from task execution.

Example

Creating a task directly (prefer using create_task utility)::

from django_simple_queue.models import Task
import json

task = Task.objects.create(
    task="myapp.tasks.send_email",
    args=json.dumps({"to": "user@example.com", "subject": "Hello"})
)

QUEUED = 0 class-attribute instance-attribute

PROGRESS = 1 class-attribute instance-attribute

COMPLETED = 2 class-attribute instance-attribute

FAILED = 3 class-attribute instance-attribute

CANCELLED = 4 class-attribute instance-attribute

STATUS_CHOICES = ((QUEUED, _('Queued')), (PROGRESS, _('In progress')), (COMPLETED, _('Completed')), (FAILED, _('Failed')), (CANCELLED, _('Cancelled'))) class-attribute instance-attribute

as_dict: dict property

Returns a dictionary representation of the task.

Useful for JSON serialization in API responses.

Returns:

Name	Type	Description
dict	dict	Task data with string-formatted dates and status display.

clean_task()

Validates that the task field contains a valid callable path.

Called automatically during model validation. Ensures the dotted path can be imported and resolved to a callable.

Raises:

Type	Description
ValidationError	If the task path is invalid or not callable.

Source code in django_simple_queue/models.py

def clean_task(self):
    """
    Validates that the task field contains a valid callable path.

    Called automatically during model validation. Ensures the dotted path
    can be imported and resolved to a callable.

    Raises:
        ValidationError: If the task path is invalid or not callable.
    """
    try:
        self._callable_task(self.task)
    except (ImportError, AttributeError, TypeError, ValueError) as e:
        raise ValidationError({
            'task': ValidationError(
                _('Invalid callable: %(error)s'),
                code='invalid',
                params={'error': str(e)}
            )
        })

clean_args()

Validates that the args field contains valid JSON.

Called automatically during model validation. Ensures the args field can be parsed as JSON (should be a dict when deserialized).

Raises:

Type	Description
ValidationError	If the args field is not valid JSON.

Source code in django_simple_queue/models.py

def clean_args(self):
    """
    Validates that the args field contains valid JSON.

    Called automatically during model validation. Ensures the args field
    can be parsed as JSON (should be a dict when deserialized).

    Raises:
        ValidationError: If the args field is not valid JSON.
    """
    if self.args is None or self.args == "":
        return

    try:
        json.loads(self.args)
    except (json.JSONDecodeError, TypeError, ValueError) as e:
        raise ValidationError({
            'args': ValidationError(
                _('Invalid JSON: %(error)s'),
                code='invalid',
                params={'error': str(e)}
            )
        })

Status Constants

Constant	Value	Description
Task.QUEUED	0	Task is waiting to be picked up
Task.PROGRESS	1	Task is currently executing
Task.COMPLETED	2	Task finished successfully
Task.FAILED	3	Task encountered an error
Task.CANCELLED	4	Task was manually cancelled

Fields

Field	Type	Description
id	UUID	Primary key (auto-generated)
created	DateTime	When the task was created
modified	DateTime	When the task was last updated
task	CharField	Dotted path to the callable
args	TextField	JSON-encoded keyword arguments
status	IntegerField	Current status (see constants above)
output	TextField	Return value from the task function
worker_pid	IntegerField	PID of the worker process
error	TextField	Error message and traceback
log	TextField	Captured stdout/stderr/logging

Usage Examples

Query Tasks by Status

from django_simple_queue.models import Task

# Get all pending tasks
pending = Task.objects.filter(status=Task.QUEUED)

# Get failed tasks from today
from django.utils import timezone
from datetime import timedelta

today = timezone.now().date()
failed_today = Task.objects.filter(
    status=Task.FAILED,
    created__date=today
)

Check Task Result

task = Task.objects.get(id=task_id)

if task.status == Task.COMPLETED:
    print(f"Result: {task.output}")
elif task.status == Task.FAILED:
    print(f"Error: {task.error}")
    print(f"Logs: {task.log}")

Re-queue a Failed Task

task = Task.objects.get(id=task_id)
task.status = Task.QUEUED
task.error = None
task.worker_pid = None
task.save()

Get JSON Representation

task = Task.objects.get(id=task_id)
data = task.as_dict
# {
#     "id": "...",
#     "created": "...",
#     "status": "Completed",
#     "output": "...",
#     ...
# }