### Setting up and Running Django-Q Tests
Source: https://github.com/koed00/django-q/blob/master/README.rst
This guide provides a step-by-step process to set up a development environment, install necessary dependencies, and execute tests for the Django-Q project. It includes instructions for creating a virtual environment, installing Python packages, and using Docker Compose to manage required services like Disque, Redis, and MongoDB for testing.
```bash
# Create virtual environment
python -m venv venv
# Install requirements
venv/bin/pip install -r requirements.txt
# Install test dependencies
venv/bin/pip install pytest pytest-django
# Install django-q
venv/bin/python setup.py develop
# Run required services (you need to have docker-compose installed)
docker-compose -f test-services-docker-compose.yaml up -d
# Run tests
venv/bin/pytest
# Stop the services required by tests (when you no longer plan to run tests)
docker-compose -f test-services-docker-compose.yaml down
```
--------------------------------
### Install Redis-py Client
Source: https://github.com/koed00/django-q/blob/master/docs/install.rst
Installs the Redis-py client library, used to interface with both Redis and Disque brokers for Django Q.
```Shell
$ pip install redis
```
--------------------------------
### Install Iron-mq Python Binding
Source: https://github.com/koed00/django-q/blob/master/docs/install.rst
Installs the official Python binding for the IronMQ broker, used with Django Q.
```Shell
$ pip install iron-mq
```
--------------------------------
### Install Pymongo for MongoDB Broker
Source: https://github.com/koed00/django-q/blob/master/docs/install.rst
Installs the Pymongo library, necessary for using MongoDB as a message broker with Django Q.
```Shell
$ pip install pymongo
```
--------------------------------
### Install Hiredis Parser
Source: https://github.com/koed00/django-q/blob/master/docs/install.rst
Installs the Hiredis C library, which provides a faster parser for Redis during high loads compared to the standard Python parser.
```Shell
$ pip install hiredis
```
--------------------------------
### Install Psutil for CPU Affinity
Source: https://github.com/koed00/django-q/blob/master/docs/install.rst
Installs the psutil library, an optional requirement that adds CPU affinity settings to the Django Q cluster.
```Shell
$ pip install psutil
```
--------------------------------
### Running Django Management Commands with Django Q
Source: https://github.com/koed00/django-q/blob/master/docs/examples.rst
This snippet shows how to execute Django management commands asynchronously using `async_task` or schedule them using `schedule`. It provides an example of calling `clearsessions` immediately or on an hourly schedule.
```Python
from django_q.tasks import async_task, schedule
async_task('django.core.management.call_command','clearsessions')
# or clear those sessions every hour
schedule('django.core.management.call_command',
'clearsessions',
schedule_type='H')
```
--------------------------------
### Install Django Q with pip
Source: https://github.com/koed00/django-q/blob/master/docs/install.rst
Installs the latest version of the Django Q library using the pip package manager.
```Shell
$ pip install django-q
```
--------------------------------
### Install Django Q Sentry Add-on
Source: https://github.com/koed00/django-q/blob/master/docs/install.rst
Installs the django-q-sentry add-on, which provides a Sentry error reporter for Django Q.
```Shell
$ pip install django-q[sentry]
```
--------------------------------
### Install Boto3 for Amazon SQS
Source: https://github.com/koed00/django-q/blob/master/docs/install.rst
Installs the Boto3 library, required for using Amazon SQS as a message broker with Django Q.
```Shell
$ pip install boto3
```
--------------------------------
### Sending Scheduled and Immediate Emails Asynchronously with Django Q
Source: https://github.com/koed00/django-q/blob/master/docs/examples.rst
This example demonstrates how to send an immediate welcome email and schedule a follow-up email one hour later using Django Q's `async_task` and `schedule` functions. It shows how to integrate email sending into background tasks to avoid blocking user requests, ensuring a responsive application.
```python
# Welcome mail with follow up example
from datetime import timedelta
from django.utils import timezone
from django_q.tasks import async_task, schedule
from django_q.models import Schedule
def welcome_mail(user):
msg = 'Welcome to our website'
# send this message right away
async_task('django.core.mail.send_mail',
'Welcome',
msg,
'from@example.com',
[user.email])
# and this follow up email in one hour
msg = 'Here are some tips to get you started...'
schedule('django.core.mail.send_mail',
'Follow up',
msg,
'from@example.com',
[user.email],
schedule_type=Schedule.ONCE,
next_run=timezone.now() + timedelta(hours=1))
# since the `repeats` defaults to -1
# this schedule will erase itself after having run
```
--------------------------------
### Install Django-Q with pip
Source: https://github.com/koed00/django-q/blob/master/README.rst
This command installs the latest version of Django-Q using pip, the Python package installer. It's the first step to integrate Django-Q into your project.
```Shell
$ pip install django-q
```
--------------------------------
### Install Django Q Rollbar Add-on
Source: https://github.com/koed00/django-q/blob/master/docs/install.rst
Installs the django-q-rollbar add-on, which provides a Rollbar error reporter for Django Q.
```Shell
$ pip install django-q[rollbar]
```
--------------------------------
### Run Django Q Cluster
Source: https://github.com/koed00/django-q/blob/master/docs/install.rst
Starts the Django Q cluster process to handle asynchronous tasks.
```Shell
$ python manage.py qcluster
```
--------------------------------
### Start Django Q Cluster Monitor
Source: https://github.com/koed00/django-q/blob/master/docs/monitor.rst
This command-line instruction starts the live monitoring interface for Django Q clusters, displaying real-time statistics and status updates.
```bash
$ python manage.py qmonitor
```
--------------------------------
### Group Example: Parzen-Window Estimation with Django Q
Source: https://github.com/koed00/django-q/blob/master/docs/examples.rst
This example demonstrates using Django Q's task grouping feature for parallel processing, specifically for Kernel density estimation using the Parzen-window technique. It defines an estimation function (`parzen_estimation`) and a `parzen_async` function that dispatches multiple tasks to a named group and then collects their results, showcasing distributed computing patterns.
```Python
import numpy
from django_q.tasks import async_task, result_group, delete_group
# the estimation function
def parzen_estimation(x_samples, point_x, h):
k_n = 0
for row in x_samples:
x_i = (point_x - row[:, numpy.newaxis]) / h
for row in x_i:
if numpy.abs(row) > (1 / 2):
break
else:
k_n += 1
return h, (k_n / len(x_samples)) / (h ** point_x.shape[1])
# create 100 calculations and return the collated result
def parzen_async():
# clear the previous results
delete_group('parzen', cached=True)
mu_vec = numpy.array([0, 0])
cov_mat = numpy.array([[1, 0], [0, 1]])
sample = numpy.random. \
multivariate_normal(mu_vec, cov_mat, 10000)
widths = numpy.linspace(1.0, 1.2, 100)
x = numpy.array([[0], [0]])
# async_task them with a group label to the cache backend
for w in widths:
async_task(parzen_estimation, sample, x, w,
group='parzen', cached=True)
# return after 100 results
return result_group('parzen', count=100, cached=True)
```
--------------------------------
### Executing Shell Commands with Django Q (Python < 3.5)
Source: https://github.com/koed00/django-q/blob/master/docs/examples.rst
This snippet demonstrates how to execute shell commands asynchronously using Django Q and Python's `subprocess` module. It shows examples of making a backup copy using `subprocess.call` and capturing output from `ls -l` using `subprocess.check_output`.
```Python
from django_q.tasks import async_task, result
# make a backup copy of setup.py
async_task('subprocess.call', ['cp', 'setup.py', 'setup.py.bak'])
# call ls -l and dump the output
task_id=async_task('subprocess.check_output', ['ls', '-l'])
# get the result
dir_list = result(task_id)
```
--------------------------------
### Install Croniter for Scheduler
Source: https://github.com/koed00/django-q/blob/master/docs/install.rst
Installs the croniter package, an optional dependency used by the Django Q scheduler to parse cron expressions.
```Shell
$ pip install croniter
```
--------------------------------
### Get One-Off Django Q Cluster Summary
Source: https://github.com/koed00/django-q/blob/master/docs/monitor.rst
Use this command to get a quick, summed summary of all active Django Q cluster statistics, including task rate and average execution time, without starting the full monitor.
```bash
$ python manage.py qinfo
```
--------------------------------
### Install Pyrollbar Error Notifier
Source: https://github.com/koed00/django-q/blob/master/docs/install.rst
Installs the Pyrollbar library, an error notifier for Rollbar, allowing management of worker errors in one place.
```Shell
$ pip install rollbar
```
--------------------------------
### Start Django-Q Cluster
Source: https://github.com/koed00/django-q/blob/master/README.rst
This management command starts a Django-Q cluster, initiating the worker pool to process tasks. It should be run in your Django project's environment.
```Shell
$ python manage.py qcluster
```
--------------------------------
### Asynchronous Report Generation with Callback Hooks in Django Q
Source: https://github.com/koed00/django-q/blob/master/docs/examples.rst
This example demonstrates how to generate a report asynchronously and handle its outcome using a hook. The `create_html_report` function generates the report, and the `email_report` hook function is called upon completion to either email the report if successful or notify administrators of a failure, providing robust error handling.
```python
# Report generation with hook example
from django_q.tasks import async_task
# views.py
# user requests a report.
def create_report(request):
async_task('tasks.create_html_report',
request.user,
hook='tasks.email_report')
```
```python
# tasks.py
from django_q.tasks import async_task
# report generator
def create_html_report(user):
html_report = 'We had a great quarter!'
return html_report
# report mailer
def email_report(task):
if task.success:
# Email the report
async_task('django.core.mail.send_mail',
'The report you requested',
task.result,
'from@example.com',
task.args[0].email)
else:
# Tell the admins something went wrong
async_task('django.core.mail.mail_admins',
'Report generation failed',
task.result)
```
--------------------------------
### Example output for Chain class execution
Source: https://github.com/koed00/django-q/blob/master/docs/chain.rst
Shows the expected output from the execution of a `Chain` instance containing `math.copysign` and `math.floor` tasks.
```python
[-1.0, 1]
```
--------------------------------
### Configure Django-Q Error Reporting with Rollbar
Source: https://github.com/koed00/django-q/blob/master/docs/configure.rst
This example demonstrates how to configure Django-Q to redirect worker exceptions to an error reporter like Rollbar. It requires installing Django-Q with necessary extras and providing specific configuration settings such as `access_token` and `environment`.
```Python
Q_CLUSTER = {
'error_reporter': {
'rollbar': {
'access_token': '32we33a92a5224jiww8982',
'environment': 'Django-Q'
}
}
}
```
--------------------------------
### Executing Shell Commands with Django Q (Python 3.5+)
Source: https://github.com/koed00/django-q/blob/master/docs/examples.rst
This snippet illustrates how to run shell commands asynchronously with Django Q using the `subprocess.run` function, introduced in Python 3.5. It demonstrates capturing return codes, arguments, and piping output using `subprocess.PIPE`.
```Python
from django_q.tasks import async_task, result
# make a backup copy of setup.py
tid = async_task('subprocess.run', ['cp', 'setup.py', 'setup.py.bak'])
# get the result
r=result(tid, 500)
# we can now look at the original arguments
# >>> r.args
# ['cp', 'setup.py', 'setup.py.bak']
# and the returncode
# >>> r.returncode
# 0
# to capture the output we'll need a pipe
from subprocess import PIPE
# call ls -l and pipe the output
tid = async_task('subprocess.run', ['ls', '-l'], stdout=PIPE)
# get the result
res = result(tid, 500)
# print the output
print(res.stdout)
```
--------------------------------
### Real-time Haystack Indexing with Django Q Signals
Source: https://github.com/koed00/django-q/blob/master/docs/examples.rst
This example demonstrates how to integrate Django Q with Django model signals to perform real-time Haystack indexing asynchronously. The first part sets up a `post_save` signal handler to dispatch an `index_object` task, and the second part defines the `index_object` function that handles the actual Haystack update, ensuring indexes are updated without blocking the main thread.
```Python
# Real time Haystack indexing
from .models import Document
from django.db.models.signals import post_save
from django.dispatch import receiver
from django_q.tasks import async_task
# hook up the post save handler
@receiver(post_save, sender=Document)
def document_changed(sender, instance, **kwargs):
async_task('tasks.index_object', sender, instance, save=False)
# turn off result saving to not flood your database
```
```Python
# tasks.py
from haystack import connection_router, connections
def index_object(sender, instance):
# get possible backends
backends = connection_router.for_write(instance=instance)
for backend in backends:
# get the index for this model
index = connections[backend].get_unified_index()\
.get_index(sender)
# update it
index.update_object(instance, using=backend)
```
--------------------------------
### API Reference: Chain.run method
Source: https://github.com/koed00/django-q/blob/master/docs/chain.rst
Starts queueing the chain to the worker cluster for execution.
```APIDOC
run()
Returns: The chain's group id.
```
--------------------------------
### Example Error Output from Synchronous Django Q Task
Source: https://github.com/koed00/django-q/blob/master/docs/tasks.rst
Provides an example of the error message that might be printed when a synchronous Django Q task fails. This specific output indicates an `ImportError`, suggesting a missing module in the executed task.
```bash
An error occurred: ImportError("No module named 'my'",)
```
--------------------------------
### Configure Django Q with Disque Broker
Source: https://github.com/koed00/django-q/blob/master/docs/configure.rst
Set a list of available Disque nodes for the cluster to connect to. Includes examples for direct node lists and using environment variables for Heroku's Tynd Disque addon.
```Python
# example disque connection
Q_CLUSTER = {
'name': 'DisqueBroker',
'workers': 4,
'timeout': 60,
'retry': 60,
'disque_nodes': ['127.0.0.1:7711', '127.0.0.1:7712']
}
```
```Python
# example Tynd Disque connection
import os
Q_CLUSTER = {
'name': 'TyndBroker',
'workers': 8,
'timeout': 30,
'retry': 60,
'bulk': 10,
'disque_nodes': os.environ['TYND_DISQUE_NODES'].split(','),
'disque_auth': os.environ['TYND_DISQUE_AUTH']
}
```
--------------------------------
### Offloading Tasks with async_task in Django Q
Source: https://github.com/koed00/django-q/blob/master/docs/tasks.rst
Demonstrates how to use `async_task` to offload functions to the Django Q Cluster. It shows examples of creating tasks, retrieving results, and integrating with hooks for post-execution processing. Results can be retrieved immediately or after a specified wait time.
```python
from django_q.tasks import async_task, result
# create the task
async_task('math.copysign', 2, -2)
# or with import and storing the id
import math.copysign
task_id = async_task(copysign, 2, -2)
# get the result
task_result = result(task_id)
# result returns None if the task has not been executed yet
# you can wait for it
task_result = result(task_id, 200)
# but in most cases you will want to use a hook:
async_task('math.modf', 2.5, hook='hooks.print_result')
# hooks.py
def print_result(task):
print(task.result)
```
--------------------------------
### Add Django-Q to Django INSTALLED_APPS
Source: https://github.com/koed00/django-q/blob/master/README.rst
After installation, add 'django_q' to your project's INSTALLED_APPS in settings.py to register it with Django. This makes Django-Q's models and management commands available.
```Python
INSTALLED_APPS = (
# other apps
'django_q',
)
```
--------------------------------
### Start Django Q Cluster using manage.py
Source: https://github.com/koed00/django-q/blob/master/docs/cluster.rst
This command initiates the Django Q cluster, which uses Python's multiprocessing module to manage worker pools for task handling. It should be run from your Django project's root directory.
```bash
$ python manage.py qcluster
```
--------------------------------
### Implementing Asynchronous Mass Notifications for User Changes
Source: https://github.com/koed00/django-q/blob/master/docs/examples.rst
These two code blocks provide different implementations for the `inform_everyone` task, which is triggered when a user's email changes. The first uses Django's `send_mass_mail` for efficiency, while the second demonstrates sending individual emails asynchronously using `async_task` for each recipient, showcasing flexibility in task design.
```python
# tasks.py
def inform_everyone(user):
mails = []
for u in User.objects.exclude(pk=user.pk):
msg = f"Dear {u.username}, {user.username} has a new email address: {user.email}"
mails.append(('New email', msg,
'from@example.com', [u.email]))
return send_mass_mail(mails)
```
```python
# or do it async again
def inform_everyone_async(user):
for u in User.objects.exclude(pk=user.pk):
msg = f"Dear {u.username}, {user.username} has a new email address: {u.email}"
async_task('django.core.mail.send_mail',
'New email', msg, 'from@example.com', [u.email])
```
--------------------------------
### Print Django Q Cluster Configuration
Source: https://github.com/koed00/django-q/blob/master/docs/monitor.rst
This command prints the current configuration settings for your Django Q clusters, useful for debugging or verifying setup.
```bash
$ python manage.py qinfo --config
```
--------------------------------
### Configure Django Q Cluster in a Procfile
Source: https://github.com/koed00/django-q/blob/master/docs/cluster.rst
For platforms like Heroku or when using Honcho, this Procfile entry defines a worker process that starts the Django Q cluster. It ensures the cluster is launched automatically upon deployment.
```Procfile
worker: python manage.py qcluster
```
--------------------------------
### Asynchronously Handling Django Model Signals with Django Q
Source: https://github.com/koed00/django-q/blob/master/docs/examples.rst
This snippet illustrates how to use Django Q to trigger asynchronous actions in response to Django model signals, specifically `pre_save`. It shows how to detect changes (e.g., a user's email address) and offload subsequent actions to a background task, preventing delays in object saving and improving application responsiveness.
```python
# Message on object change
from django.contrib.auth.models import User
from django.db.models.signals import pre_save
from django.dispatch import receiver
from django_q.tasks import async_task
# set up the pre_save signal for our user
@receiver(pre_save, sender=User)
def email_changed(sender, instance, **kwargs):
try:
user = sender.objects.get(pk=instance.pk)
except sender.DoesNotExist:
pass # new user
else:
# has his email changed?
if not user.email == instance.email:
# tell everyone
async_task('tasks.inform_everyone', instance)
```
--------------------------------
### Python Project Dependency List with Hashes
Source: https://github.com/koed00/django-q/blob/master/requirements.txt
This snippet provides a list of Python package dependencies, their required versions, platform/Python version compatibility markers, and SHA256 hashes for secure and reproducible builds. Each line specifies a package, its version, and conditional installation requirements, followed by multiple hashes for verification.
```Python
ansicon==1.89.0; platform_system == "Windows" and python_version >= "2.7" \
--hash=sha256:f1def52d17f65c2c9682cf8370c03f541f410c1752d6a14029f97318e4b9dfec \
--hash=sha256:e4d039def5768a47e4afec8e89e83ec3ae5a26bf00ad851f914d1240b444d2b1
arrow==1.1.1; python_version >= "3.6" \
--hash=sha256:77a60a4db5766d900a2085ce9074c5c7b8e2c99afeaa98ad627637ff6f292510 \
--hash=sha256:dee7602f6c60e3ec510095b5e301441bc56288cb8f51def14dcb3079f623823a
asgiref==3.3.4; python_version >= "3.6" \
--hash=sha256:92906c611ce6c967347bbfea733f13d6313901d54dcca88195eaeb52b2a8e8ee \
--hash=sha256:d1216dfbdfb63826470995d31caed36225dcaf34f182e0fa257a4dd9e86f1b78
blessed==1.18.1; python_version >= "2.7" \
--hash=sha256:dd7c0d33db9a2e7f597b446996484d0ed46e1586239db064fb5025008937dcae \
--hash=sha256:8b09936def6bc06583db99b65636b980075733e13550cb6af262ce724a55da23
django-picklefield==3.0.1; python_version >= "3" \
--hash=sha256:15ccba592ca953b9edf9532e64640329cd47b136b7f8f10f2939caa5f9ce4287 \
--hash=sha256:3c702a54fde2d322fe5b2f39b8f78d9f655b8f77944ab26f703be6c0ed335a35
django==3.2.4; python_version >= "3.6" \
--hash=sha256:ea735cbbbb3b2fba6d4da4784a0043d84c67c92f1fdf15ad6db69900e792c10f \
--hash=sha256:66c9d8db8cc6fe938a28b7887c1596e42d522e27618562517cc8929eb7e7f296
jinxed==1.1.0; platform_system == "Windows" and python_version >= "2.7" \
--hash=sha256:6a61ccf963c16aa885304f27e6e5693783676897cea0c7f223270c8b8e78baf8 \
--hash=sha256:d8f1731f134e9e6b04d95095845ae6c10eb15cb223a5f0cabdea87d4a279c305
python-dateutil==2.8.1; python_version >= "3.6" and python_full_version < "3.0.0" or python_full_version >= "3.3.0" and python_version >= "3.6" \
--hash=sha256:73ebfe9dbf22e832286dafa60473e4cd239f8592f699aa5adaf10050e6e1823c \
--hash=sha256:75bb3f31ea686f1197762692a9ee6a7550b59fc6ca3a1f4b5d7e32fb98e2da2a
pytz==2021.1; python_version >= "3.6" \
--hash=sha256:eb10ce3e7736052ed3623d49975ce333bcd712c7bb19a58b9e2089d4057d0798 \
--hash=sha256:83a4a90894bf38e243cf052c8b58f381bfe9a7a483f6a9cab140bc7f702ac4da
six==1.16.0; python_version >= "3.6" and python_full_version < "3.0.0" or python_full_version >= "3.3.0" and python_version >= "3.6" \
--hash=sha256:8abb2f1d86890a2dfb989f9a77cfcfd3e47c2a354b01111771326f8aa26e0254 \
--hash=sha256:1e61c37477a1626458e36f7b1d82aa5c9b094fa4802892072e49de9c60c4c926
sqlparse==0.4.1; python_version >= "3.6" \
--hash=sha256:017cde379adbd6a1f15a61873f43e8274179378e95ef3fede90b5aa64d304ed0 \
--hash=sha256:0f91fd2e829c44362cbcfab3e9ae12e22badaa8a29ad5ff599f9ec109f0454e8
typing-extensions==3.10.0.0; python_version < "3.8" and python_version >= "3.6" \
--hash=sha256:0ac0f89795dd19de6b97debb0c6af1c70987fd80a2d62d1958f7e56fcc31b497 \
--hash=sha256:779383f6086d90c99ae41cf0ff39aac8a7937a9283ce0a414e5dd782f4c94a84 \
--hash=sha256:50b6f157849174217d0656f99dc82fe932884fb250826c18350e159ec6cdf342
wcwidth==0.2.5; python_version >= "2.7" \
--hash=sha256:beb4802a9cebb9144e99086eff703a642a13d6a0052920003a230f3294bbe784 \
--hash=sha256:c4d647b99872929fdb7bdcaa4fbe7f01413ed3d98077df798530e5b04f116c83
```
--------------------------------
### Subscribe to Django Q Lifecycle Signals in Python
Source: https://github.com/koed00/django-q/blob/master/docs/signals.rst
This Python example demonstrates how to connect to Django Q's lifecycle signals (pre_enqueue, pre_execute, post_execute) using Django's `receiver` decorator. It shows how to define callback functions that print information about the task at different stages of its execution.
```Python
from django.dispatch import receiver
from django_q.signals import pre_enqueue, pre_execute, post_execute
@receiver(pre_enqueue)
def my_pre_enqueue_callback(sender, task, **kwargs):
print(f"Task {task['name']} will be queued")
@receiver(pre_execute)
def my_pre_execute_callback(sender, func, task, **kwargs):
print(f"Task {task['name']} will be executed by calling {func}")
@receiver(post_execute)
def my_post_execute_callback(sender, task, **kwargs):
print(f"Task {task['name']} was executed with result {task['result']}")
```
--------------------------------
### Create Asynchronous Tasks with Django-Q
Source: https://github.com/koed00/django-q/blob/master/README.rst
This Python code demonstrates how to offload tasks asynchronously using `async_task`. It shows examples of passing functions by name or reference, retrieving results, and using result hooks for post-execution processing. Results are `None` until the task completes, and can be waited for.
```Python
from django_q.tasks import async_task, result
# create the task
async_task('math.copysign', 2, -2)
# or with a reference
import math.copysign
task_id = async_task(copysign, 2, -2)
# get the result
task_result = result(task_id)
# result returns None if the task has not been executed yet
# you can wait for it
task_result = result(task_id, 200)
# but in most cases you will want to use a hook:
async_task('math.modf', 2.5, hook='hooks.print_result')
# hooks.py
def print_result(task):
print(task.result)
```
--------------------------------
### Example Output for result_group 'modf'
Source: https://github.com/koed00/django-q/blob/master/docs/group.rst
This snippet shows the expected output when retrieving results from the 'modf' group using `result_group` after the tasks have completed.
```python
[(0.0, 0.0), (0.0, 1.0), (0.0, 2.0), (0.0, 3.0)]
```
--------------------------------
### Check Django-Q Overall Statistics
Source: https://github.com/koed00/django-q/blob/master/README.rst
This command provides overall statistics about your Django-Q setup, including task counts and other relevant metrics. It's useful for a quick health check.
```Shell
$ python manage.py qinfo
```
--------------------------------
### Configure Django Q with django-redis Cache Backend
Source: https://github.com/koed00/django-q/blob/master/docs/configure.rst
Leverage an existing django-redis cache connection as the broker for Django Q. This example shows how to specify the name of the cache connection to use.
```Python
# example django-redis connection
Q_CLUSTER = {
'name': 'DJRedis',
'workers': 4,
'timeout': 90,
'django_redis': 'default'
}
```
--------------------------------
### Configure Django-Q ORM Broker with Replica Database
Source: https://github.com/koed00/django-q/blob/master/docs/configure.rst
This configuration snippet demonstrates how to set up the Django ORM backend for Django-Q to work correctly with a replica database. By setting `has_replica` to `True`, Django-Q ensures proper operation with a multi-database setup and a Database Router.
```Python
Q_CLUSTER = {
...
'orm': 'default',
'has_replica': True
}
```
--------------------------------
### Perform Asynchronous Parzen Estimation with Django-Q
Source: https://github.com/koed00/django-q/blob/master/docs/examples.rst
This Python function demonstrates how to perform a Parzen estimation asynchronously using Django-Q's `async_iter`. It processes an iterable of arguments in parallel, calculates a result ID, and retrieves the cached result with a specified timeout. Requires `numpy` for statistical operations.
```python
def parzen_async():
mu_vec = numpy.array([0, 0])
cov_mat = numpy.array([[1, 0], [0, 1]])
sample = numpy.random. \
multivariate_normal(mu_vec, cov_mat, 10000)
widths = numpy.linspace(1.0, 1.2, 100)
x = numpy.array([[0], [0]])
# async_task them with async_task iterable
args = [(sample, x, w) for w in widths]
result_id = async_iter(parzen_estimation, args, cached=True)
# return the cached result or timeout after 10 seconds
return result(result_id, wait=10000, cached=True)
```
--------------------------------
### Illustrate Django-Q CPU Affinity Behavior
Source: https://github.com/koed00/django-q/blob/master/docs/configure.rst
This example visually explains how the `cpu_affinity` setting affects worker distribution across available processors in a Django-Q cluster. It shows how workers are assigned to specific CPUs based on the affinity value, optimizing performance for high-traffic scenarios.
```Python
4 processors, 4 workers, cpu_affinity: 1
worker 1 cpu [0]
worker 2 cpu [1]
worker 3 cpu [2]
worker 4 cpu [3]
4 processors, 4 workers, cpu_affinity: 2
```
--------------------------------
### Create a Python HTTP Health Check Server for Django-Q Clusters
Source: https://github.com/koed00/django-q/blob/master/docs/examples.rst
This Python script implements a simple HTTP server (localhost:8080) to validate the health status of Django-Q clusters. It checks if all clusters are in an `IDLE` or `WORKING` state and returns a 200 OK response if healthy, otherwise a 500 error. The server requires Django-Q's cache to be enabled and the Django project's settings module to be correctly configured. It can be run using `python worker_hc.py`.
```python
from http.server import BaseHTTPRequestHandler, HTTPServer
from mtt_app.settings.base import EMAIL_USE_TLS
import os
import django
# Set the correct path to you settings module
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "my.settings.path")
# All django stuff has to come after the setup:
django.setup()
from django_q.monitor import Stat
from django_q.conf import Conf
# Set host and port settings
hostName = "localhost"
serverPort = 8080
class HealthCheckServer(BaseHTTPRequestHandler):
def do_GET(self):
# Count the clusters and their status
happy_clusters = 0
total_clusters = 0
for stat in Stat.get_all():
total_clusters += 1
if stat.status in [Conf.IDLE, Conf.WORKING]:
happy_clusters += 1
# Return 200 response if there is at least 1 cluster running,
# and make sure all running clusters are happy
if total_clusters and happy_clusters == total_clusters:
response_code = 200
else:
response_code = 500
self.send_response(response_code)
self.send_header("Content-type", "text/html")
self.end_headers()
self.wfile.write(
bytes("Django-Q Heath Check", "utf-8")
)
self.wfile.write(
bytes(f"Health check returned {response_code} response
", "utf-8")
)
self.wfile.write(
bytes(
f"{happy_clusters} of {total_clusters} cluster(s) are happy
",
"utf-8",
)
)
if __name__ == "__main__":
webServer = HTTPServer((hostName, serverPort), HealthCheckServer)
print("Server started at http://%s:%s" % (hostName, serverPort))
try:
webServer.serve_forever()
except KeyboardInterrupt:
pass
webServer.server_close()
print("Server stopped.")
```
--------------------------------
### Django Q Cluster Class API Reference
Source: https://github.com/koed00/django-q/blob/master/docs/cluster.rst
This section provides a reference for the `Cluster` class in `django_q`, detailing its methods and attributes for programmatic interaction. It includes methods for starting, stopping, and checking the status of the cluster, along with attributes like process ID, host, and various event flags.
```APIDOC
Cluster Class:
Methods:
start(): Spawns a cluster and then returns.
stop(): Initiates stop_procedure and waits for it to finish.
stat(): Returns a Stat object with the current cluster status.
Attributes:
pid: The cluster process id.
host: The current hostname.
sentinel: Returns the multiprocessing.Process containing the sentinel.
timeout: The clusters timeout setting in seconds.
start_event: A multiprocessing.Event indicating if the sentinel has finished starting the cluster.
stop_event: A multiprocessing.Event used to instruct the sentinel to initiate the stop_procedure.
is_starting: Bool. Indicating that the cluster is busy starting up.
is_running: Bool. Tells you if the cluster is up and running.
```
--------------------------------
### Configure Django Q Cluster with Supervisor Process Manager
Source: https://github.com/koed00/django-q/blob/master/docs/cluster.rst
This `supervisor.conf` example demonstrates how to set up Supervisor to manage the Django Q cluster. The `stopasgroup` option is crucial to ensure that all child processes of the cluster are properly terminated when Supervisor stops or restarts the program.
```ini
[program:django-q]
command = python manage.py qcluster
stopasgroup = true
```
--------------------------------
### Configure Django-Q with MongoDB Message Broker
Source: https://github.com/koed00/django-q/blob/master/docs/configure.rst
This example shows how to configure Django-Q to use MongoDB as its message broker. The `mongo` dictionary accepts parameters directly from `pymongo.MongoClient`, allowing specification of host, port, or a full MongoDB URI.
```Python
Q_CLUSTER = {
'name': 'MongoDB',
'workers': 8,
'timeout': 60,
'retry': 70,
'queue_limit': 100,
'mongo': {
'host': '127.0.0.1',
'port': 27017
}
}
```
--------------------------------
### Group Asynchronous Tasks and Retrieve Results with result_group in Django-Q
Source: https://github.com/koed00/django-q/blob/master/docs/group.rst
This example demonstrates how to use `async_task` to assign multiple tasks to a named group and then use `result_group` to wait for and retrieve all results from that group. It shows a basic loop to create tasks and then block until all results are available.
```python
# result group example
from django_q.tasks import async_task, result_group
for i in range(4):
async_task('math.modf', i, group='modf')
# wait until the group has 4 results
result = result_group('modf', count=4)
print(result)
```
--------------------------------
### Illustrate Django Q Retry Mechanism with async_task
Source: https://github.com/koed00/django-q/blob/master/docs/configure.rst
This example demonstrates how the 'retry' setting in Q_CLUSTER interacts with task execution, especially for long-running tasks. It shows how a task might be re-queued and executed multiple times by different workers if the retry time is less than the task's execution duration, potentially leading to multiple executions of the same task and resource contention.
```python
# settings.py
Q_CLUSTER = {
'retry': 5,
'workers': 4,
'orm': 'default',
}
# example.py
from django_q.tasks import async_task
async_task('time.sleep', 22)
```
--------------------------------
### Manage Task Groups via AsyncTask Object Methods in Django-Q
Source: https://github.com/koed00/django-q/blob/master/docs/group.rst
This example demonstrates how to use group-related methods directly on an `AsyncTask` object. It shows creating an `AsyncTask` with a group identifier and then using `a.result_group()` to wait for a specified number of results within that task's group.
```python
from django_q.tasks import AsyncTask
# add a task to the math group and run it cached
a = AsyncTask('math.floor', 2.5, group='math', cached=True)
# wait until this tasks group has 10 results
result = a.result_group(count=10)
```
--------------------------------
### Configure Django Q Cluster with Circus Process Manager
Source: https://github.com/koed00/django-q/blob/master/docs/cluster.rst
This `circus.ini` example shows how to configure Circus to manage a single instance of the Django Q cluster. It sets up basic Circus parameters and defines a watcher for the `qcluster` command, ensuring the cluster runs as a supervised process.
```ini
[circus]
check_delay = 5
endpoint = tcp://127.0.0.1:5555
pubsub_endpoint = tcp://127.0.0.1:5556
stats_endpoint = tcp://127.0.0.1:5557
[watcher:django_q]
cmd = python manage.py qcluster
numprocesses = 1
copy_env = True
```
--------------------------------
### Run Django Migrations for Django Q
Source: https://github.com/koed00/django-q/blob/master/docs/install.rst
Executes Django migrations to create the necessary database tables for Django Q.
```Shell
$ python manage.py migrate
```
--------------------------------
### Configure Django Q in INSTALLED_APPS
Source: https://github.com/koed00/django-q/blob/master/docs/install.rst
Adds 'django_q' to the INSTALLED_APPS tuple in your Django project's settings.py file to enable the application.
```Python
INSTALLED_APPS = (
# other apps
'django_q',
)
```
--------------------------------
### Instantiating and Running Tasks with AsyncTask Class
Source: https://github.com/koed00/django-q/blob/master/docs/tasks.rst
Shows how to use the `AsyncTask` class to encapsulate a task and its options within a single object. It demonstrates instantiation, modifying parameters like `args` and `cached`, running the task, and retrieving results with optional wait times. Re-running is required after parameter changes.
```python
# AsyncTask class instance example
from django_q.tasks import AsyncTask
# instantiate an async task
a = AsyncTask('math.floor', 1.5, group='math')
# you can set or change keywords afterwards
a.cached = True
# run it
a.run()
# wait indefinitely for the result and print it
print(a.result(wait=-1))
# change the args
a.args = (2.5,)
# run it again
a.run()
# wait max 10 seconds for the result and print it
print(a.result(wait=10))
```
--------------------------------
### Run Django Migrations for Django-Q
Source: https://github.com/koed00/django-q/blob/master/README.rst
Execute Django migrations to create the necessary database tables for Django-Q. This step initializes the database schema required for task management and scheduling.
```Shell
$ python manage.py migrate
```
--------------------------------
### API Reference: Chain.length method
Source: https://github.com/koed00/django-q/blob/master/docs/chain.rst
Gets the total number of tasks in the chain.
```APIDOC
length()
Returns: Length of the chain (int).
```
--------------------------------
### API Reference: Chain.current method
Source: https://github.com/koed00/django-q/blob/master/docs/chain.rst
Gets the index of the currently executing element within the chain.
```APIDOC
current()
Returns: Current chain index (int).
```
--------------------------------
### Creating Django-Q Schedules with Python
Source: https://github.com/koed00/django-q/blob/master/docs/schedules.rst
This snippet demonstrates various ways to create scheduled tasks in Django-Q. It shows how to use the `schedule` wrapper function for different scheduling types like daily, hourly, minutes, and cron expressions, as well as direct object creation using the `Schedule` model. It also covers specifying `q_options` and assigning tasks to specific clusters.
```python
# Use the schedule wrapper
from django_q.tasks import schedule
schedule('math.copysign',
2, -2,
hook='hooks.print_result',
schedule_type='D')
# Or create the object directly
from django_q.models import Schedule
Schedule.objects.create(func='math.copysign',
hook='hooks.print_result',
args='2,-2',
schedule_type=Schedule.DAILY
)
# In case you want to use q_options
# Specify the broker by using the property broker_name in q_options
schedule('math.sqrt',
9,
hook='hooks.print_result',
q_options={'timeout': 30, 'broker_name': 'broker_1'},
schedule_type=Schedule.HOURLY)
# Run a schedule every 5 minutes, starting at 6 today
# for 2 hours
import arrow
schedule('math.hypot',
3, 4,
schedule_type=Schedule.MINUTES,
minutes=5,
repeats=24,
next_run=arrow.utcnow().replace(hour=18, minute=0))
# Use a cron expression
schedule('math.hypot',
3, 4,
schedule_type=Schedule.CRON,
cron = '0 22 * * 1-5')
# Restrain a schedule to a specific cluster
schedule('math.hypot',
3, 4,
schedule_type=Schedule.DAILY,
cluster='my_cluster')
```
--------------------------------
### Django-Q Schedule Class API Reference
Source: https://github.com/koed00/django-q/blob/master/docs/schedules.rst
Detailed API documentation for the `Schedule` database model, outlining its attributes for defining task parameters and scheduling types, and its methods for task management.
```APIDOC
class Schedule:
description: A database model for task schedules.
Attributes:
id: Primary key
name: A name for your schedule. Tasks created by this schedule will assume this or the primary key as their group id.
func: The function to be scheduled
hook: Optional hook function to be called after execution.
args: Positional arguments for the function.
kwargs: Keyword arguments for the function
schedule_type: The type of schedule. Follows Schedule.TYPE
TYPE: ONCE, MINUTES, HOURLY, DAILY, WEEKLY, MONTHLY, QUARTERLY, YEARLY, CRON
minutes: The number of minutes the MINUTES schedule should use. Is ignored for other schedule types.
cron: A cron string describing the schedule. You need the optional `croniter` package installed for this.
repeats: Number of times to repeat the schedule. -1=Always, 0=Never, n =n. When set to -1, this will keep counting down.
cluster: Task will be executed only on a cluster with a matching name.
next_run: Datetime of the next scheduled execution.
task: Id of the last task generated by this schedule.
ONCE: 'O' the schedule will only run once. If it has a negative repeats it will be deleted after it has run. If you want to keep the result, set repeats to a positive number.
MINUTES: 'I' will run every minutes after its first run.
HOURLY: 'H' the scheduled task will run every hour after its first run.
DAILY: 'D' the scheduled task will run every day at the time of its first run.
WEEKLY: 'W' the task will run every week on they day and time of the first run.
MONTHLY: 'M' the tasks runs every month on they day and time of the last run. Months are tricky. If you schedule something on the 31st of the month and the next month has only 30 days or less, the task will run on the last day of the next month. It will however continue to run on that day, e.g. the 28th, in subsequent months.
QUARTERLY: 'Q' this task runs once every 3 months on the day and time of the last run.
YEARLY: 'Y' only runs once a year. The same caution as with months apply; If you set this to february 29th, it will run on february 28th in the following years.
CRON: 'C' uses the optional `croniter` package to determine a schedule based on a cron expression.
Methods:
last_run(): Admin link to the last executed task.
success(): Returns the success status of the last executed task.
```
--------------------------------
### Reusing Broker Connections for Django Q Tasks
Source: https://github.com/koed00/django-q/blob/master/docs/tasks.rst
Demonstrates how to optimize broker connection usage by initializing a broker instance once and passing it to multiple `async_task` calls. This reduces overhead and prevents running out of connections when making many individual task requests.
```python
# broker connection economy example
from django_q.tasks import async_task
from django_q.brokers import get_broker
broker = get_broker()
for i in range(50):
async_task('math.modf', 2.5, broker=broker)
```
--------------------------------
### Manage sequential tasks with Chain class
Source: https://github.com/koed00/django-q/blob/master/docs/chain.rst
Illustrates the use of the `Chain` class for a more convenient way to manage sequential tasks. It shows how to create a cached chain, append individual tasks, run the chain, and retrieve its results.
```python
# Chain async
from django_q.tasks import Chain
# create a chain that uses the cache backend
chain = Chain(cached=True)
# add some tasks
chain.append('math.copysign', 1, -1)
chain.append('math.floor', 1)
# run it
chain.run()
print(chain.result())
```
--------------------------------
### Disque Broker for Django Q
Source: https://github.com/koed00/django-q/blob/master/docs/brokers.rst
Disque offers guaranteed delivery through message receipts and is comparable in speed to Redis. It supports bulk task retrieval and requires Django's Cache framework for monitoring. It's still considered Alpha software.
```APIDOC
Disque:
- Delivery receipts
- Atomic
- Needs Django's `Cache framework` configured for monitoring
- Compatible with `Tynd` Disque addon on Heroku
- Still considered Alpha software
- Supports bulk dequeue
- Requires `Redis-py` client library: `pip install redis`
- See the `disque_configuration` configuration section for more info.
```
--------------------------------
### Django Q Cluster Configuration: catch_up
Source: https://github.com/koed00/django-q/blob/master/docs/configure.rst
The default behavior for schedules that didn't run while a cluster was down, is to play catch up and execute all the missed time slots until things are back on schedule. You can override this behavior by setting catch_up to False. This will make those schedules run only once when the cluster starts and normal scheduling resumes.
```APIDOC
Parameter: catch_up
Type: boolean
Default: True
```
--------------------------------
### Retrieve and Filter Task Groups with fetch_group and count_group in Django-Q
Source: https://github.com/koed00/django-q/blob/master/docs/group.rst
This example demonstrates how to use `fetch_group` to retrieve a Django QuerySet of `Task` objects, allowing for more flexible filtering (e.g., excluding failures). It also shows `count_group` for counting failures and highlights equivalences with `result_group`'s default behavior.
```python
# fetch group example
from django_q.tasks import fetch_group, count_group, result_group
# count the number of failures
failure_count = count_group('modf', failures=True)
# only use the successes
results = fetch_group('modf')
if failure_count:
results = results.exclude(success=False)
results = [task.result for task in successes]
# this is the same as
results = fetch_group('modf', failures=False)
results = [task.result for task in successes]
# and the same as
results = result_group('modf') # filters failures by default
```