django_dramatiq
django_dramatiq is a Django app that integrates with Dramatiq.
Requirements
Example
You can find an example application built with django_dramatiq here.
Installation
pip install django-dramatiq
Add django_dramatiq to installed apps before any of your custom apps:
import os
INSTALLED_APPS = [
"django_dramatiq",
"myprojectapp1",
"myprojectapp2",
# etc...
]
Configure your broker in settings.py:
DRAMATIQ_BROKER = {
"BROKER": "dramatiq.brokers.rabbitmq.RabbitmqBroker",
"OPTIONS": {
"url": "amqp://localhost:5672",
},
"MIDDLEWARE": [
"dramatiq.middleware.Prometheus",
"dramatiq.middleware.AgeLimit",
"dramatiq.middleware.TimeLimit",
"dramatiq.middleware.Callbacks",
"dramatiq.middleware.Retries",
"django_dramatiq.middleware.DbConnectionsMiddleware",
"django_dramatiq.middleware.AdminMiddleware",
]
}
# Defines which database should be used to persist Task objects when the
# AdminMiddleware is enabled. The default value is "default".
DRAMATIQ_TASKS_DATABASE = "default"
You may also configure a result backend:
DRAMATIQ_RESULT_BACKEND = {
"BACKEND": "dramatiq.results.backends.redis.RedisBackend",
"BACKEND_OPTIONS": {
"url": "redis://localhost:6379",
},
"MIDDLEWARE_OPTIONS": {
"result_ttl": 60000
}
}
Usage
Declaring tasks
django_dramatiq will auto-discover tasks defined in tasks modules in each of your installed apps. For example, if you have an app named customers, your tasks for that app should live in a module called customers.tasks:
import dramatiq
from django.core.mail import send_mail
from .models import Customer
@dramatiq.actor
def email_customer(customer_id, subject, message):
customer = Customer.get(pk=customer_id)
send_mail(subject, message, "[email protected]", [customer.email])
Running workers
django_dramatiq comes with a management command you can use to auto-discover task modules and run workers:
python manage.py rundramatiq
If your project for some reason has apps with modules named tasks that are not intended for use with Dramatiq, you can ignore them:
DRAMATIQ_IGNORED_MODULES = (
'app1.tasks',
'app2.tasks',
'app3.tasks.utils',
'app3.tasks.utils.*',
...
)
The wildcard detection will ignore all sub modules from that point on. You will need to ignore the module itself if you don't want the __init__.py to be processed.
Testing
You should have a separate settings file for test. In that file, overwrite the broker to use Dramatiq's StubBroker:
DRAMATIQ_BROKER = {
"BROKER": "dramatiq.brokers.stub.StubBroker",
"OPTIONS": {},
"MIDDLEWARE": [
"dramatiq.middleware.AgeLimit",
"dramatiq.middleware.TimeLimit",
"dramatiq.middleware.Callbacks",
"dramatiq.middleware.Pipelines",
"dramatiq.middleware.Retries",
"django_dramatiq.middleware.DbConnectionsMiddleware",
"django_dramatiq.middleware.AdminMiddleware",
]
}
Using pytest-django
In your conftest module set up fixtures for your broker and a worker:
import dramatiq
import pytest
@pytest.fixture
def broker():
broker = dramatiq.get_broker()
broker.flush_all()
return broker
@pytest.fixture
def worker(broker):
worker = dramatiq.Worker(broker, worker_timeout=100)
worker.start()
yield worker
worker.stop()
In your tests, use those fixtures whenever you want background tasks to be executed:
def test_customers_can_be_emailed(transactional_db, broker, worker, mailoutbox):
customer = Customer(email="[email protected]")
# Assuming "send_welcome_email" enqueues an "email_customer" task
customer.send_welcome_email()
# Wait for all the tasks to be processed
broker.join("default")
worker.join()
assert len(mailoutbox) == 1
assert mailoutbox[0].subject == "Welcome Jim!"
Using unittest
A simple test case has been provided that will automatically set up the broker and worker for each test, which are accessible as attributes on the test case. Note that DramatiqTestCase inherits django.test.TransactionTestCase.
from django.core import mail
from django.test import override_settings
from django_dramatiq.test import DramatiqTestCase
class CustomerTestCase(DramatiqTestCase):
@override_settings(EMAIL_BACKEND='django.core.mail.backends.locmem.EmailBackend')
def test_customers_can_be_emailed(self):
customer = Customer(email="[email protected]")
# Assuming "send_welcome_email" enqueues an "email_customer" task
customer.send_welcome_email()
# Wait for all the tasks to be processed
self.broker.join(customer.queue_name)
self.worker.join()
self.assertEqual(len(mail.outbox), 1)
self.assertEqual(mail.outbox[0].subject, "Welcome Jim!")
Cleaning up old tasks
The AdminMiddleware stores task metadata in a relational DB so it's a good idea to garbage collect that data every once in a while. You can use the delete_old_tasks actor to achieve this on a cron:
delete_old_tasks.send(max_task_age=86400)
Middleware
- django_dramatiq.middleware.DbConnectionsMiddleware
- This middleware is vital in taking care of closing expired connections after each message is processed.
- django_dramatiq.middleware.AdminMiddleware
- This middleware stores metadata about tasks in flight to a database and exposes them via the Django admin.
Custom keyword arguments to Middleware
Some middleware classes require dynamic arguments. An example of this would be the backend argument to dramatiq.middleware.GroupCallbacks.
To do this, you might add the middleware to your settings.py:
DRAMATIQ_BROKER = {
...
"MIDDLEWARE": [
...
"dramatiq.middleware.GroupCallbacks",
...
]
...
}
Next, you need to extend DjangoDramatiqConfig to provide the arguments for this middleware:
from django_dramatiq.apps import DjangoDramatiqConfig
class CustomDjangoDramatiqConfig(DjangoDramatiqConfig):
@classmethod
def middleware_groupcallbacks_kwargs(cls):
return {"rate_limiter_backend": cls.get_rate_limiter_backend()}
CustomDjangoDramatiqConfig.initialize()
Notice the naming convention, to provide arguments to dramatiq.middleware.GroupCallbacks you need to add a @classmethod with the name middleware_<middleware_name>_kwargs, where <middleware_name> is the lowercase name of the middleware.
Finally, add the custom app config to your settings.py, replacing the existing django_dramatiq app config:
INSTALLED_APPS = [
...
"yourapp.apps.CustomDjangoDramatiqConfig",
...
]
Usage with django-configurations
To use django_dramatiq together with django-configurations you need to define your own rundramatiq command as a subclass of the one in this package.
In YOURPACKAGE/management/commands/rundramatiq.py:
from django_dramatiq.management.commands.rundramatiq import Command as RunDramatiqCommand
class Command(RunDramatiqCommand):
def discover_tasks_modules(self):
tasks_modules = super().discover_tasks_modules()
tasks_modules[0] = "YOURPACKAGE.dramatiq_setup"
return tasks_modules
And in YOURPACKAGE/dramatiq_setup.py:
import django
from configurations.importer import install
install(check_options=True)
django.setup()
Running project tests locally
Install the dev dependencies with pip install -e '.[dev]' and then run tox.
License
django_dramatiq is licensed under Apache 2.0. Please see LICENSE for licensing details.
1.6k Dec 28, 2022
486 Jan 08, 2023
1 Dec 16, 2022
371 Jan 01, 2023
8 Nov 17, 2022
2 Feb 15, 2022
7 Nov 07, 2021
24 Dec 14, 2022
30 Oct 24, 2022
117 Dec 30, 2022
2.5k Jan 03, 2023
3 Oct 01, 2022
1 Dec 22, 2021
130 Dec 17, 2022
7 Dec 05, 2022
20.7k Jan 01, 2023
10 May 13, 2022
364 Jan 02, 2023