diff --git a/django_q/brokers/__init__.py b/django_q/brokers/__init__.py index 959d392..32eb6fa 100644 --- a/django_q/brokers/__init__.py +++ b/django_q/brokers/__init__.py @@ -1,6 +1,9 @@ -from django_q.conf import Conf +import importlib + from django.core.cache import caches, InvalidCacheBackendError +from django_q.conf import Conf + class Broker(object): def __init__(self, list_key=Conf.PREFIX): @@ -155,19 +158,29 @@ def get_broker(list_key=Conf.PREFIX): :type list_key: str :return: """ + # custom + if Conf.BROKER_CLASS: + module, func = Conf.BROKER_CLASS.rsplit('.', 1) + m = importlib.import_module(module) + broker = getattr(m, func) + return broker(list_key=list) # disque - if Conf.DISQUE_NODES: + elif Conf.DISQUE_NODES: from brokers import disque return disque.Disque(list_key=list_key) + # Iron MQ elif Conf.IRON_MQ: from brokers import ironmq return ironmq.IronMQBroker(list_key=list_key) + # SQS elif Conf.SQS: from brokers import aws_sqs return aws_sqs.Sqs(list_key=list_key) + # ORM elif Conf.ORM: from brokers import orm return orm.ORM(list_key=list_key) + # Mongo elif Conf.MONGO: from brokers import mongo return mongo.Mongo(list_key=list_key) diff --git a/django_q/conf.py b/django_q/conf.py index 58989d4..f5cb7df 100644 --- a/django_q/conf.py +++ b/django_q/conf.py @@ -1,4 +1,5 @@ import logging +from copy import deepcopy from signal import signal from multiprocessing import cpu_count, Queue @@ -50,6 +51,9 @@ class Conf(object): # ORM broker ORM = conf.get('orm', None) + # Custom broker class + BROKER_CLASS = conf.get('broker_class', None) + # Database Poll POLL = conf.get('poll', 0.2) @@ -169,7 +173,7 @@ if not logger.handlers: # rollbar if Conf.ROLLBAR: - rollbar_conf = Conf.ROLLBAR + rollbar_conf = deepcopy(Conf.ROLLBAR) try: import rollbar rollbar.init(rollbar_conf.pop('access_token'), environment=rollbar_conf.pop('environment'), **rollbar_conf) @@ -180,9 +184,6 @@ else: rollbar = None - - - # get parent pid compatibility def get_ppid(): if hasattr(os, 'getppid'): diff --git a/django_q/tests/test_brokers.py b/django_q/tests/test_brokers.py index c0d6b93..8c63a98 100644 --- a/django_q/tests/test_brokers.py +++ b/django_q/tests/test_brokers.py @@ -1,9 +1,11 @@ -from time import sleep -import pytest import os +from time import sleep + +import pytest import redis -from django_q.conf import Conf + from django_q.brokers import get_broker, Broker +from django_q.conf import Conf from django_q.humanhash import uuid @@ -47,6 +49,15 @@ def test_redis(): Conf.DJANGO_REDIS = 'default' +def test_custom(): + Conf.BROKER_CLASS = 'brokers.redis_broker.Redis' + broker = get_broker() + assert broker.ping() is True + assert broker.info() is not None + assert broker.__class__.__name__ == 'Redis' + Conf.BROKER_CLASS = None + + def test_disque(): Conf.DISQUE_NODES = ['127.0.0.1:7711'] # check broker @@ -132,15 +143,15 @@ def test_ironmq(): broker.acknowledge(task[0]) assert broker.dequeue() is None # Retry test - #Conf.RETRY = 1 - #broker.enqueue('test') - #assert broker.dequeue() is not None - #sleep(3) + # Conf.RETRY = 1 + # broker.enqueue('test') # assert broker.dequeue() is not None - #task = broker.dequeue()[0] - #assert len(task) > 0 - #broker.acknowledge(task[0]) - #sleep(3) + # sleep(3) + # assert broker.dequeue() is not None + # task = broker.dequeue()[0] + # assert len(task) > 0 + # broker.acknowledge(task[0]) + # sleep(3) # delete job task_id = broker.enqueue('test') broker.delete(task_id) diff --git a/docs/brokers.rst b/docs/brokers.rst index 60604a2..7345e98 100644 --- a/docs/brokers.rst +++ b/docs/brokers.rst @@ -103,6 +103,36 @@ However for a medium message rate and scheduled tasks, this is the most convenie * Queue editable in Django Admin * See the :ref:`orm_configuration` configuration on how to set it up. + + +Custom Broker +------------- +You can override the :class:`Broker` or any of its existing derived broker types. + +.. code-block:: python + + # example Custom broker.py + from django_q.brokers import Broker + + class CustomBroker(Broker): + def info(self): + return 'My Custom Broker' + +Using the :ref:`broker_class` configuration setting you can then instruct Django Q to use this instead of one of the existing brokers: + +.. code-block:: python + + # example Custom broker class connection + + Q_CLUSTER = { + 'name': 'Custom', + 'workers': 8, + 'timeout': 60, + 'broker_class: 'myapp.broker.CustomBroker' + } + +If you do write a custom broker for one of the many message queueing servers out there we don't support yet, please consider contributing it to the project. + Reference --------- The :class:`Broker` class is used internally to communicate with the different types of brokers. diff --git a/docs/configure.rst b/docs/configure.rst index 56453e5..5e063c5 100644 --- a/docs/configure.rst +++ b/docs/configure.rst @@ -292,6 +292,23 @@ mongo_db When using the MongoDB broker you can optionally provide a database name to use for the queues. Defaults to default database if available, otherwise ``django-q`` +.. _broker_class: + +broker_class +~~~~~~~~~~~~ +You can use a custom broker class for your cluster workers:: + + # example Custom broker class connection + + Q_CLUSTER = { + 'name': 'Custom', + 'workers': 8, + 'timeout': 60, + 'broker_class: 'myapp.broker.CustomBroker' + } + +Make sure your ``CustomBroker`` class inherits from either the base :class:`Broker` class or one of its children. + .. _bulk: bulk