Django Q

Django Q is a native Django task queue, scheduler, and worker application that utilizes Python multiprocessing. It is designed to handle asynchronous tasks, scheduled/cron/repeated tasks, and provides robust monitoring and integration capabilities.

Key features include:

• Worker Pools: Uses multiprocessing worker pools.
• Task Management: Supports asynchronous tasks, scheduled tasks, cron jobs, and repeated tasks.
• Reliability: Supports signed and compressed packages, and tracks failures/successes in a database or cache.
• Task Orchestration: Supports result hooks, groups, and chains.
• Broker Support: Compatible with Redis, Disque, IronMQ, SQS, MongoDB, or the Django ORM.
• Integrations: Django Admin integration, Rollbar and Sentry support, and PaaS compatibility for multiple instances.
• Monitoring: Multi-cluster monitor support.

Django Q integrates with the standard Django admin interface using proxy models. When you access the Django Q admin pages, you will find three primary models for monitoring and managing tasks:

1. Successful tasks: Displays tasks that executed without errors. You can view details, delete them, or filter by group (schedule name or group ID). The table is searchable by name, func, and group.
2. Failed tasks: Displays tasks that encountered errors. The error details are typically stored in the result field. You can resubmit a failed task to the queue directly from the admin action menu.
3. Scheduled tasks: Allows you to create, edit, delete, or monitor tasks set to run at specific intervals.

Note: If you have configured save_limit to prevent saving successful tasks, you will only see failed results in the admin.

You can run multiple clusters on different machines to work on the same queue. For this to work correctly, you must ensure the following requirements are met:

1. They must connect to the same broker.
2. They must use the same cluster name (configured in settings).
3. They must share the same Django SECRET_KEY.

Django Q provides mechanisms to handle multiple related tasks as a single unit.

Using Task Groups

1. Assign a group name to multiple async_task calls.
2. Use result_group(group_name, count=N) to wait for and retrieve the results once a specific number of tasks in that group have completed.
3. Use delete_group(group_name) to clear previous results from the cache.

Using async_iter

async_iter is a high-level abstraction that automatically utilizes the cache backend and groups to return a single result from an iterable of arguments. It is often cleaner than manually managing groups for batch processing.

If you are using the orm_broker broker, an additional admin view for Queued tasks is enabled. This view shows all task packages currently waiting in the broker queue.

• Lock column: Indicates when the package was picked up by the cluster; this is used to determine if a task has expired.
• Management: You can edit or delete queued tasks directly from this view, which is useful for development.

Django Q uses a watchdog mechanism to handle hung workers:

1. Before executing a task, a Worker sets a countdown timer with the Sentinel.
2. The timer is reset after the task execution completes.
3. The Sentinel continuously checks these timers. If a timer reaches zero (indicating a worker is stuck or unresponsive), the Sentinel terminates that worker and reincarnates a new one.

The Schedule.repeats attribute controls how many times a scheduled task runs.

• Finite runs: Set a positive integer (e.g., 24) to run the task a specific number of times. The count decrements each time the schedule runs.
• Indefinite runs: Set repeats to -1. The schedule will run indefinitely, and the count will continue to decrement (e.g., -1, -2, -3...), serving as an execution counter.
• Pausing: Set repeats to 0 to pause a schedule.
• Schedule.ONCE type:
  • A positive number keeps the schedule but prevents it from running again.
  • A negative number causes the schedule to be deleted from the database.
  • To re-run a ONCE schedule that was paused (repeats set to 0), change the repeats to a non-zero value and set a new run time.

To create a custom error reporting plugin for Django Q, you must provide a class that satisfies the following requirements:

1. Constructor: The class must accept keyword arguments (**kwargs). These arguments are sourced directly from the Q_CLUSTER configuration in settings.py.
2. report method: The class must implement a report method. This method is automatically called by the Django Q cluster whenever an error occurs.

Example structure for a plugin class:

class MyCustomReporter:
    def __init__(self, **kwargs):
        # Settings from Q_CLUSTER are available here
        self.api_key = kwargs.get('api_key')

    def report(self, exception, task, cluster):
        # Logic to send the exception to your service
        pass