Django AMQP

AMQP support for Django, with implementations for RabbitMQ and Azure ServiceBus.

Overview

Django AMQP is an extension for Django that provides Advanced Message Queuing Protocol (AMQP) support. It enables you to easily implement message queues and task processing in your Django applications using either RabbitMQ or Azure Service Bus as the backend.

The library integrates with Django's task framework, allowing you to define, queue, and process asynchronous tasks with support for scheduled/deferred execution.

Installation

pip install django-amqp

OR

uv add django-amqp

Configuration

Add django_amqp to your INSTALLED_APPS in settings.py:

INSTALLED_APPS = [
    # ...
    'django_amqp',
    # ...
]

Configure your task backend in settings.py:

SERVICEBUS_CONNECTION_STRING = os.environ.get("SERVICEBUS_CONNECTION_STRING")

TASKS = {
    "default": {
        "BACKEND": "django_amqp.backend.ServiceBusBackend",
        "QUEUES": [
            "default",
        ],
        "OPTIONS": {"connection_string": SERVICEBUS_CONNECTION_STRING},
    }
}

Usage

Defining Tasks

Create a task using Django's task decorator:

from django.tasks import task

@task(queue_name="my-queue")
def my_background_task(param1, param2, **kwargs):
    # Your task logic here
    pass

NOTE: queue_name defaults to "default" (as per django tasks implementation). You may want to name your main worker ServiceBus/RabbitMQ queue as "default" to match this. Otherwise, specifyqueue_name for each task.*

Queueing Tasks

Queue a Task for Immediate Execution

from myapp.tasks import my_background_task

# Queue for immediate execution
my_background_task.enqueue(param1="value1", param2="value2")

you can also batch queue

from myapp.tasks import my_background_task

# Batch Queue for immediate execution
my_background_task.get_backend().batch_enqueue(
    my_background_task, [
            ([], {param1="value1", param2="value2"}),
            ([], {param1="value3", param2="value4"})
        ]
)
# It's more performant to queue many messages at once, instead of making a connection
# for each

Queue and Cancel a Task for Delayed Execution

You can cancel scheduled messages using the sequence number returned when scheduling a message:

from django_amqp.service_bus import cancel_deferred_message

# Schedule a task and get the sequence number
sequence_number = my_background_task.delay_until(future_time, param1="value1")

# Later, if you need to cancel it
cancel_deferred_message(sequence_number, queue_name="my-queue")

Running Workers

To process messages from a specific queue:

python manage.py amqp_worker --queue-name="my-queue" --burst

The --burst flag tells the worker to process all available messages and then exit.

Error Handling

When using the ServiceBus Backend, failed tasks are automatically moved to the dead-letter queue with the error reason and description. You can implement additional error handling in your task functions.

License

This project is licensed under the BSD 3-Clause License - see the LICENSE file for details.