Utilities

Utility functions for creating and managing tasks.

API Reference

utils

TaskNotAllowedError

Bases: Exception

Raised when attempting to create a task that is not in the allowed list.

create_task(task: str, args: dict) -> uuid.UUID

Create a new task to be executed by the worker.

Parameters:

Name	Type	Description	Default
task	str	Dotted path to the callable (e.g., "myapp.tasks.process_order")	required
args	dict	Dictionary of keyword arguments to pass to the callable	required

Returns:

Type	Description
UUID	UUID of the created task

Raises:

Type	Description
TypeError	If args is not a dict
TaskNotAllowedError	If task is not in DJANGO_SIMPLE_QUEUE_ALLOWED_TASKS

Example

from django_simple_queue.utils import create_task

task_id = create_task( task="myapp.tasks.send_email", args={"to": "user@example.com", "subject": "Hello"} )

Source code in django_simple_queue/utils.py

def create_task(task: str, args: dict) -> uuid.UUID:
    """
    Create a new task to be executed by the worker.

    Args:
        task: Dotted path to the callable (e.g., "myapp.tasks.process_order")
        args: Dictionary of keyword arguments to pass to the callable

    Returns:
        UUID of the created task

    Raises:
        TypeError: If args is not a dict
        TaskNotAllowedError: If task is not in DJANGO_SIMPLE_QUEUE_ALLOWED_TASKS

    Example:
        from django_simple_queue.utils import create_task

        task_id = create_task(
            task="myapp.tasks.send_email",
            args={"to": "user@example.com", "subject": "Hello"}
        )
    """
    if not isinstance(args, dict):
        raise TypeError("args should be of type dict.")

    if not is_task_allowed(task):
        allowed = get_allowed_tasks()
        if allowed is not None:
            raise TaskNotAllowedError(
                f"Task '{task}' is not in the allowed list. "
                f"Add it to DJANGO_SIMPLE_QUEUE_ALLOWED_TASKS in settings.py"
            )

    obj = Task.objects.create(
        task=task,
        args=json.dumps(args)
    )
    return obj.id

create_task

The primary way to enqueue tasks for background execution.

Signature

def create_task(task: str, args: dict) -> UUID:
    ...

Parameters

Parameter	Type	Description
task	str	Dotted path to the callable (e.g., "myapp.tasks.send_email")
args	dict	Keyword arguments to pass to the callable

Returns

• UUID: The unique identifier of the created task

Raises

• TypeError: If args is not a dictionary
• TaskNotAllowedError: If task is not in DJANGO_SIMPLE_QUEUE_ALLOWED_TASKS

Examples

from django_simple_queue.utils import create_task

# Basic usage
task_id = create_task(
    task="myapp.tasks.send_email",
    args={"to": "user@example.com", "subject": "Hello"}
)

# With complex arguments
task_id = create_task(
    task="myapp.tasks.process_order",
    args={
        "order_id": 12345,
        "options": {
            "notify": True,
            "priority": "high"
        }
    }
)

# Check the created task
from django_simple_queue.models import Task
task = Task.objects.get(id=task_id)
print(task.status)  # 0 (QUEUED)

TaskNotAllowedError

Exception raised when attempting to create a task that is not in the allowlist.

Example

from django_simple_queue.utils import create_task, TaskNotAllowedError

try:
    task_id = create_task(
        task="some.unknown.function",
        args={}
    )
except TaskNotAllowedError as e:
    print(f"Task not allowed: {e}")

Best Practices

Pass IDs, Not Objects

# Good
create_task(
    task="myapp.tasks.process_user",
    args={"user_id": user.id}
)

# Bad - can't serialize model instances
create_task(
    task="myapp.tasks.process_user",
    args={"user": user}  # TypeError!
)

Use JSON-Serializable Arguments

# Good
create_task(
    task="myapp.tasks.schedule",
    args={"date": "2024-01-15T10:00:00"}
)

# Bad - datetime not JSON-serializable by default
create_task(
    task="myapp.tasks.schedule",
    args={"date": datetime.now()}  # Error!
)