Merge pull request #27 from Koed00/dev

Adds `save` override options for tasks
2026-09-16 22:17:55 +08:00 · 2015-07-21 03:55:57 -07:00
parent 61b859b046 ad3a76be4f
commit 9e331740cd
14 changed files with 131 additions and 55 deletions
--- a/MANIFEST.in
+++ b/MANIFEST.in
@@ -2,5 +2,4 @@ include LICENSE
 include README.rst
 include django_q/management/*.py
 include django_q/management/commands/*.py
-exclude django_q/management/commands/qtest.py
 include django_q/migrations/*.py
--- a/django_q/init.py
+++ b/django_q/init.py
@@ -7,6 +7,7 @@ sys.path.insert(0, myPath)
 from .tasks import async, schedule, result, result_group, fetch, fetch_group, count_group, delete_group
 from .models import Task, Schedule, Success, Failure
 from .cluster import Cluster
+from .monitor import Stat

 VERSION = (0, 4, 0)

--- a/django_q/cluster.py
+++ b/django_q/cluster.py
@@ -394,12 +394,11 @@ def save_task(task):
    Saves the task package to Django
    """
    # SAVE LIMIT < 0 : Don't save success
-    if Conf.SAVE_LIMIT < 0 and task['success']:
+    if not task.get('save', Conf.SAVE_LIMIT > 0) and task['success']:
        return
    # SAVE LIMIT > 0: Prune database, SAVE_LIMIT 0: No pruning
    if task['success'] and 0 < Conf.SAVE_LIMIT < Success.objects.count():
        Success.objects.last().delete()
-
    try:
        Task.objects.create(id=task['id'],
                            name=task['name'],
@@ -435,8 +434,9 @@ def scheduler(list_key=Conf.Q_LIST):
            # single value won't eval to tuple, so:
            if type(args) != tuple:
                args = (args,)
+        q_options = kwargs.get('q_options', {})
        if s.hook:
-            kwargs['hook'] = s.hook
+            q_options['hook'] = s.hook
        # set up the next run time
        if not s.schedule_type == s.ONCE:
            next_run = arrow.get(s.next_run)
@@ -455,8 +455,9 @@ def scheduler(list_key=Conf.Q_LIST):
            s.next_run = next_run.datetime
            s.repeats += -1
        # send it to the cluster
-        kwargs['list_key'] = list_key
-        kwargs['group'] = s.name or s.id
+        q_options['list_key'] = list_key
+        q_options['group'] = s.name or s.id
+        kwargs['q_options'] = q_options
        s.task = tasks.async(s.func, *args, **kwargs)
        # log it
        if not s.task:
--- a/django_q/tasks.py
+++ b/django_q/tasks.py
@@ -20,16 +20,14 @@ def async(func, *args, **kwargs):
    """
    Sends a task to the cluster
    """
-    # optional hook
-    hook = kwargs.pop('hook', None)
-    # optional list_key
-    list_key = kwargs.pop('list_key', Conf.Q_LIST)
-    # optional redis connection
-    redis = kwargs.pop('redis', redis_client)
-    # optional sync mode
-    sync = kwargs.pop('sync', False)
-    # optional group
-    group = kwargs.pop('group', None)
+    # get options from q_options dict or direct from kwargs
+    options = kwargs.pop('q_options', kwargs)
+    hook = options.pop('hook', None)
+    list_key = options.pop('list_key', Conf.Q_LIST)
+    redis = options.pop('redis', redis_client)
+    sync = options.pop('sync', False)
+    group = options.pop('group', None)
+    save = options.pop('save', None)
    # get an id
    tag = uuid()
    # build the task package
@@ -40,6 +38,8 @@ def async(func, *args, **kwargs):
        task['hook'] = hook
    if group:
        task['group'] = group
+    if save is not None:
+        task['save'] = save
    # sign it
    pack = signing.SignedPackage.dumps(task)
    if sync:
--- a/django_q/tests/test_cluster.py
+++ b/django_q/tests/test_cluster.py
@@ -124,6 +124,9 @@ def test_async(r, admin_user):
    h = async('django_q.tests.tasks.word_multiply', 2, word='django', hook='fail.me', list_key=list_key, redis=r)
    # args unpickle test
    j = async('django_q.tests.tasks.get_user_id', admin_user, list_key=list_key, group='test_j', redis=r)
+    # q_options and save opt_out test
+    k = async('django_q.tests.tasks.get_user_id', admin_user,
+              q_options={'list_key': list_key, 'group': 'test_k', 'redis': r, 'save': False, 'timeout': 90})
    # check if everything has a task id
    assert isinstance(a, str)
    assert isinstance(b, str)
@@ -134,8 +137,9 @@ def test_async(r, admin_user):
    assert isinstance(g, str)
    assert isinstance(h, str)
    assert isinstance(j, str)
+    assert isinstance(k, str)
    # run the cluster to execute the tasks
-    task_count = 9
+    task_count = 10
    assert r.llen(list_key) == task_count
    task_queue = Queue()
    stop_event = Event()
@@ -210,7 +214,8 @@ def test_async(r, admin_user):
    assert count_group('test_j', failures=True) == 0
    assert delete_group('test_j') == 1
    assert delete_group('test_j', tasks=True) is None
-
+    # task k should not have been saved
+    assert fetch(k) is None
    r.delete(list_key)


--- a/docs/admin.rst
+++ b/docs/admin.rst
@@ -41,14 +41,14 @@ Repeats
 If you want a schedule to only run a finite amount of times, e.g. every hour for the next 24 hours, you can do that using the :attr:`Schedule.repeats` attribute.
 In this case you would set the schedule type to :attr:`Schedule.HOURLY` and the repeats to `24`. Every time the schedule runs the repeats count down until it hits zero and schedule is no longer run.

-When you set repeats to `-1` the schedule will continue indefinitely and the repeats will still count down. This can be used as an indicator of how many times the schedule has been executed.
+When you set repeats to ``-1`` the schedule will continue indefinitely and the repeats will still count down. This can be used as an indicator of how many times the schedule has been executed.

 An exception to this are schedules of type :attr:`Schedule.ONCE`. Negative repeats for this schedule type will cause it to be deleted from the database.
 This behavior is useful if you have many delayed actions which you do not necessarily need a result for. A positive number will keep the ONCE schedule, but it will not run again.

 .. note::

-    To run a `Once` schedule again, change the repeats to something other than `0`. Set a new run time before you do this or let it execute immediately.
+    To run a ``ONCE`` schedule again, change the repeats to something other than `0`. Set a new run time before you do this or let it execute immediately.


 Next run
--- a/docs/cluster.rst
+++ b/docs/cluster.rst
@@ -4,7 +4,7 @@ Cluster
 .. py:currentmodule:: django_q

 Django Q uses Python's multiprocessing module to manage a pool of workers that will handle your tasks.
-Start your cluster using Django's `manage.py` command::
+Start your cluster using Django's ``manage.py`` command::

    $ python manage.py qcluster

@@ -26,7 +26,7 @@ You should see the cluster starting ::
    10:57:40 [Q] INFO Q Cluster-31781 running.


-Stopping the cluster with ctrl-c or either the `SIGTERM` and `SIGKILL` signals, will initiate the :ref:`stop_procedure`::
+Stopping the cluster with ctrl-c or either the ``SIGTERM`` and ``SIGKILL`` signals, will initiate the :ref:`stop_procedure`::

    16:44:12 [Q] INFO Q Cluster-31781 stopping.
    16:44:12 [Q] INFO Process-1 stopping cluster processes
@@ -50,7 +50,7 @@ You can have multiple clusters on multiple machines, working on the same queue a

 - They connect to the same Redis server.
 - They use the same cluster name. See :ref:`configuration`
- They share the same `SECRET_KEY`
+- They share the same ``SECRET_KEY``

 Using a Procfile
 ----------------
@@ -80,7 +80,7 @@ An example :file:`circus.ini` ::


 Note that we only start one process. It is not a good idea to run multiple instances of the cluster in the same environment since this does nothing to increase performance and in all likelihood will diminish it.
-Control your cluster using the `workers`, `recycle` and `timeout` settings in your :ref:`configuration`
+Control your cluster using the ``workers``, ``recycle`` and ``timeout`` settings in your :ref:`configuration`

 Architecture
 ------------
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -72,7 +72,7 @@ author = 'Ilan Steemers'
 # The short X.Y version.
 version = '0.4'
 # The full version, including alpha/beta/rc tags.
-release = '0.4.0'
+release = '0.4.1'

 # The language for content autogenerated by Sphinx. Refer to documentation
 # for a list of supported languages.
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -16,7 +16,7 @@ Features
 -  Scheduled and repeated tasks
 -  Encrypted and compressed packages
 -  Failure and success database
-  Result hooks
+-  Result hooks and groups
 -  Django Admin integration
 -  PaaS compatible with multiple instances
 -  Multi cluster monitor
--- a/docs/install.rst
+++ b/docs/install.rst
@@ -6,7 +6,7 @@ Installation
    $ pip install django-q


-  Add :mod:`django_q` to `INSTALLED_APPS` in your projects :file:`settings.py`::
+-  Add :mod:`django_q` to ``INSTALLED_APPS`` in your projects :file:`settings.py`::

       INSTALLED_APPS = (
           # other apps
@@ -18,14 +18,14 @@ Installation
    $ python manage.py migrate

 -  Make sure you have a `Redis <http://redis.io/>`__ server running
-   somewhere
+   somewhere and know how to connect to it.

 .. _configuration:

 Configuration
 -------------

-Configuration is handled via the `Q_CLUSTER` dictionary in your :file:`settings.py`
+Configuration is handled via the ``Q_CLUSTER`` dictionary in your :file:`settings.py`

 .. code:: python

@@ -133,7 +133,7 @@ of the cache connection you want to use::


 .. tip::
-    Django Q uses your `SECRET_KEY` to encrypt task packages and prevent task crossover. So make sure you have it set up in your Django settings.
+    Django Q uses your ``SECRET_KEY`` to encrypt task packages and prevent task crossover. So make sure you have it set up in your Django settings.

 cpu_affinity
 ~~~~~~~~~~~~
@@ -175,7 +175,7 @@ As a rule of thumb; cpu_affinity 1 favors repetitive short running tasks, while

 .. note::

-    The `cpu_affinity` setting requires the optional :ref:`psutil <psutil>` module.
+    The ``cpu_affinity`` setting requires the optional :ref:`psutil <psutil>` module.

 Requirements
 ------------
@@ -185,7 +185,7 @@ Django Q is tested for Python 2.7 and 3.4
 -  `Django <https://www.djangoproject.com>`__

    Django Q aims to use as much of Django's standard offerings as possible
-    The code is tested against Django version `1.7.8` and `1.8.2`.
+    The code is tested against Django version `1.7.9` and `1.8.3`.

 -  `Django-picklefield <https://github.com/gintas/django-picklefield>`__

@@ -207,6 +207,7 @@ Django Q is tested for Python 2.7 and 3.4

    Django Q uses Redis as a centralized hub between your Django instances and your Q clusters.

+
 Optional
 ~~~~~~~~
 .. _psutil:
--- a/docs/schedules.rst
+++ b/docs/schedules.rst
@@ -25,6 +25,13 @@ You can manage them through the :ref:`admin_page` or directly from your code wit
                            schedule_type=Schedule.DAILY
                            )

+    # In case you want to use async options
+    schedule('math.sqrt',
+             9,
+             hook='hooks.print_result',
+             q_options={'timeout': 30},
+             schedule_type=Schedule.HOURLY)
+

 Management Commands
 -------------------
@@ -59,6 +66,7 @@ Reference
    :param str schedule_type: (O)nce, (H)ourly, (D)aily, (W)eekly, (M)onthly, (Q)uarterly, (Y)early or :attr:`Schedule.TYPE`
    :param int repeats: Number of times to repeat schedule. -1=Always, 0=Never, n =n.
    :param datetime next_run: Next or first scheduled execution datetime.
+    :param dict q_options: async options to use for this schedule
    :param kwargs: optional keyword arguments for the scheduled function.

 .. class:: Schedule
--- a/docs/tasks.rst
+++ b/docs/tasks.rst
@@ -2,6 +2,8 @@ Tasks
 =====
 .. py:currentmodule:: django_q

+.. _async:
+
 Async
 -----

@@ -31,11 +33,53 @@ Use :func:`async` from your code to quickly offload tasks to the :class:`Cluster
    def print_result(task):
        print(task.result)

+:func:`async` can take the following optional keyword arguments:
+
+hook
+""""
+The function to call after the task has been executed. This function gets passed the complete :class:`Task` object as its argument.
+
+group
+"""""
+A group label. Check :ref:`groups` for group functions.
+
+save
+""""
+Overrides the result backend's save setting for this task.
+
+timeout
+"""""""
+Overrides the cluster's timeout setting for this task.
+
+sync
+""""
+Simulates a task execution synchronously. Useful for testing.
+
+redis
+"""""
+A redis connection. In case you want to control your own connections.
+
+q_options
+"""""""""
+None of the option keywords get passed on to the task function.
+As an alternative you can also put them in
+a single keyword dict named ``q_options``. This enables you to use these keywords for your function call::
+
+    # Async options in a dict
+
+    opts = {'hook': 'hooks.print_result',
+            'group': 'math',
+            'timeout': 30}
+
+    async('math.modf', 2.5, q_options=opts)
+
+Please not that this will override any other option keywords.
+
 .. _groups:

 Groups
 ------
-You can group together results by passing :func:`async` the optional `group` keyword:
+You can group together results by passing :func:`async` the optional ``group`` keyword:

 .. code-block:: python

@@ -67,7 +111,7 @@ Instead of :func:`result_group` you can also use :func:`fetch_group` to return a
    # only use the successes
    results = fetch_group('modf')
    if failure_count:
-        results.exclude(success=False)
+        results = results.exclude(success=False)
    results =  [task.result for task in successes]

    # this is the same as
@@ -82,13 +126,13 @@ Getting results by using :func:`result_group` is of course much faster than usin

 .. note::

-   Although :func:`fetch_group` returns a queryset, due to the nature of the PickleField , calling `Queryset.values` on it will return a list of encoded results.
+   Although :func:`fetch_group` returns a queryset, due to the nature of the PickleField , calling ``Queryset.values`` on it will return a list of encoded results.
   Use list comprehension or an iterator instead.

 Synchronous testing
 -------------------

-:func:`async` can be instructed to execute a task immediately by setting the optional keyword `sync=True`.
+:func:`async` can be instructed to execute a task immediately by setting the optional keyword ``sync=True``.
 The task will then be injected straight into a worker and the result saved by a monitor instance::

    from django_q import async, fetch
@@ -113,7 +157,7 @@ Connection pooling
 ------------------

 Django Q tries to pass redis connections around its parts as much as possible to save you from running out of connections.
-When you are making individual calls to :func:`async` a lot though, it can help to set up a redis connection to pass to :func:`async`:
+When you are making individual calls to :func:`async` a lot though, it can help to set up a redis connection to reuse for :func:`async`:

 .. code:: python

@@ -133,7 +177,7 @@ Reference
 ---------

 .. py:function:: async(func, *args, hook=None, group=None, timeout=None,\
-    sync=False, redis=None, **kwargs)
+    sync=False, redis=None, q_options=None, **kwargs)

    Puts a task in the cluster queue

@@ -144,6 +188,7 @@ Reference
   :param int timeout: Overrides global cluster :ref:`timeout`.
   :param bool sync: If set to True, async will simulate a task execution
   :param redis: Optional redis connection
+   :param dict q_options: Options dict, overrides option keywords
   :param dict kwargs: Keyword arguments for the task function
   :returns: The uuid of the task
   :rtype: str
@@ -172,7 +217,7 @@ Reference
    Returns the results of a task group

    :param str group_id: the group identifier
-    :param bool failures: set this to `True` to include failed results
+    :param bool failures: set this to ``True`` to include failed results
    :returns: a list of results
    :rtype: list

@@ -181,7 +226,7 @@ Reference
    Returns a list of tasks in a group

    :param str group_id: the group identifier
-    :param bool failures: set this to `False` to exclude failed tasks
+    :param bool failures: set this to ``False`` to exclude failed tasks
    :returns: a list of Tasks
    :rtype: list

@@ -190,7 +235,7 @@ Reference
    Counts the number of task results in a group.

    :param str group_id: the group identifier
-    :param bool failures: counts the number of failures if `True`
+    :param bool failures: counts the number of failures if ``True``
    :returns: the number of tasks or failures in a group
    :rtype: int

@@ -199,7 +244,7 @@ Reference
    Deletes a group label from the database.

    :param str group_id: the group identifier
-    :param bool tasks: also deletes the associated tasks if `True`
+    :param bool tasks: also deletes the associated tasks if ``True``
    :returns: the numbers of tasks affected
    :rtype: int

@@ -218,7 +263,7 @@ Reference
        .. note::

            This is for convenience and can be used as a parameter for most functions that take a `task_id`.
-            Keep in mind however that it is not guaranteed to be unique if you store very large amounts of tasks in the database.
+            Keep in mind that it is not guaranteed to be unique if you store very large amounts of tasks in the database.

    .. py:attribute:: func

@@ -269,7 +314,7 @@ Reference
    .. py:classmethod:: get_result_group(group_id, failures=False)

    Returns a list of results from a task group.
-    Set failures to `True` to include failed results.
+    Set failures to ``True`` to include failed results.

    .. py:classmethod:: get_task(task_id)

@@ -278,22 +323,22 @@ Reference
    .. py:classmethod:: get_task_group(group_id, failures=True)

    Gets a queryset of tasks with this group id.
-    Set failures to `False` to exclude failed tasks.
+    Set failures to ``False`` to exclude failed tasks.

    .. py:classmethod::  get_group_count(group_id, failures=False)

    Returns a count of the number of tasks results in a group.
-    Returns the number of failures when `failures=True`
+    Returns the number of failures when ``failures=True``

    .. py:classmethod:: delete_group(group_id, objects=False)

    Deletes a group label only, by default.
-    If `objects=True` it will also delete the tasks in this group from the database.
+    If ``objects=True`` it will also delete the tasks in this group from the database.

 .. py:class:: Success

-    A proxy model of :class:`Task` with the queryset filtered on :attr:`Task.success` is True.
+    A proxy model of :class:`Task` with the queryset filtered on :attr:`Task.success` is ``True``.

 .. py:class:: Failure

-     A proxy model of :class:`Task` with the queryset filtered on :attr:`Task.success` is False.
+     A proxy model of :class:`Task` with the queryset filtered on :attr:`Task.success` is ``False``.
--- a/requirements.in
+++ b/requirements.in
@@ -0,0 +1,9 @@
+arrow
+blessed
+django-picklefield
+Django
+future
+hiredis
+redis
+psutil
+django-redis
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,12 +1,19 @@
+#
+# This file is autogenerated by pip-compile
+# Make changes in requirements.in, then run this to update:
+#
+#    pip-compile requirements.in
+#
 arrow==0.6.0
 blessed==1.9.5
 django-picklefield==0.3.1
-Django>=1.7.8
+django-redis==4.2.0
+django==1.8.3
 future==0.14.3
 hiredis==0.2.0
-redis==2.10.3
+msgpack-python==0.4.6     # via django-redis
 psutil==3.1.1
-django-redis==4.2.0
-
-
-
+python-dateutil==2.4.2    # via arrow
+redis==2.10.3
+six==1.9.0                # via django-picklefield, python-dateutil
+wcwidth==0.1.4            # via blessed