diff --git a/README.rst b/README.rst index e4b9578..d9e17ba 100644 --- a/README.rst +++ b/README.rst @@ -15,7 +15,7 @@ Features - Asynchronous tasks - Scheduled and repeated tasks - Encrypted and compressed packages -- Failure and success database +- Failure and success database or cache - Result hooks and groups - Django Admin integration - PaaS compatible with multiple instances diff --git a/django_q/cluster.py b/django_q/cluster.py index 343c5ca..4ca18f3 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -325,7 +325,11 @@ def monitor(result_queue, broker=None): if ack_id: broker.acknowledge(ack_id) # save the result - save_task(task) + if task.get('cached', False): + save_cached(task, broker) + else: + save_task(task) + # log the result if task['success']: logger.info(_("Processed [{}]").format(task['name'])) else: @@ -384,7 +388,7 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT): def save_task(task): """ - Saves the task package to Django + Saves the task package to Django or the cache """ # SAVE LIMIT < 0 : Don't save success if not task.get('save', Conf.SAVE_LIMIT > 0) and task['success']: @@ -409,6 +413,43 @@ def save_task(task): logger.error(e) +def save_cached(task, broker): + task_key = '{}:{}'.format(broker.list_key, task['id']) + timeout = task['cached'] + if timeout is True: + timeout = None + try: + group = task.get('group', False) + iter_count = task.get('iter_count', None) + # if it's a group append to the group list + if group: + task_key = '{}:{}:{}'.format(broker.list_key, group, task['id']) + group_key = '{}:{}:keys'.format(broker.list_key, group) + group_list = broker.cache.get(group_key) or [] + # if it's an inter group, check if we are ready + if iter_count and len(group_list) == iter_count-1: + group_args = '{}:{}:args'.format(broker.list_key, group) + # collate the results into a Task result + results = [signing.SignedPackage.loads(broker.cache.get(k))['result'] for k in group_list] + results.append(task['result']) + task['result'] = results + task['id'] = group + task['args'] = signing.SignedPackage.loads(broker.cache.get(group_args)) + save_task(task) + broker.cache.delete_many(group_list) + broker.cache.delete_many([group_key, group_args]) + return + # save the group list + group_list.append(task_key) + broker.cache.set(group_key, group_list) + # save the task + broker.cache.set(task_key, + signing.SignedPackage.dumps(task), + timeout) + except Exception as e: + logger.error(e) + + def scheduler(broker=None): """ Creates a task from a schedule at the scheduled time and schedules next run diff --git a/django_q/conf.py b/django_q/conf.py index 9ca2146..ccfd080 100644 --- a/django_q/conf.py +++ b/django_q/conf.py @@ -113,6 +113,10 @@ class Conf(object): # The Django cache to use CACHE = conf.get('cache', 'default') + # Use the cache as result backend. Can be 'True' or an integer representing the global cache timeout. + # i.e 'cached: 60' , will make all results go the cache and expire in 60 seconds. + CACHED = conf.get('cached', False) + # If set to False the scheduler won't execute tasks in the past. # Instead it will run once and reschedule the next run in the future. Defaults to True. CATCH_UP = conf.get('catch_up', True) diff --git a/django_q/tasks.py b/django_q/tasks.py index 62da8f1..17ae5b1 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -1,4 +1,4 @@ -"""Provides task functionalities.""" +"""Provides task functionality.""" from multiprocessing import Queue, Value # django @@ -15,7 +15,7 @@ from django_q.brokers import get_broker def async(func, *args, **kwargs): - """Send a task to the cluster.""" + """Queue a task for the cluster.""" # get options from q_options dict or direct from kwargs options = kwargs.pop('q_options', kwargs) hook = options.pop('hook', None) @@ -23,6 +23,8 @@ def async(func, *args, **kwargs): sync = options.pop('sync', False) group = options.pop('group', None) save = options.pop('save', None) + cached = options.pop('cached', Conf.CACHED) + iter_count = options.pop('iter_count', None) # get an id tag = uuid() # build the task package @@ -38,6 +40,10 @@ def async(func, *args, **kwargs): task['group'] = group if save is not None: task['save'] = save + if cached: + task['cached'] = cached + if iter_count: + task['iter_count'] = iter_count # sign it pack = signing.SignedPackage.dumps(task) if sync or Conf.SYNC: @@ -83,7 +89,7 @@ def schedule(func, *args, **kwargs): ) -def result(task_id, wait=0): +def result(task_id, wait=0, cached=Conf.CACHED): """ Return the result of the named task. @@ -91,9 +97,12 @@ def result(task_id, wait=0): :param task_id: the task name or uuid :type wait: int :param wait: number of milliseconds to wait for a result + :param bool cached: run this against the cache backend :return: the result object of this task :rtype: object """ + if cached: + return result_cached(task_id, wait) start = time.time() while True: r = Task.get_result(task_id) @@ -104,18 +113,76 @@ def result(task_id, wait=0): time.sleep(0.01) -def result_group(group_id, failures=False): +def result_cached(task_id, wait=0, broker=None): + """ + Return the result from the cache backend + """ + if not broker: + broker = get_broker() + start = time.time() + while True: + r = broker.cache.get('{}:{}'.format(broker.list_key, task_id)) + if r: + return signing.SignedPackage.loads(r)['result'] + if (time.time() - start) * 1000 >= wait: + break + time.sleep(0.01) + + +def result_group(group_id, failures=False, wait=0, count=None, cached=Conf.CACHED): """ Return a list of results for a task group. :param str group_id: the group id :param bool failures: set to True to include failures + :param int count: Block until there are this many results in the group + :param bool cached: run this against the cache backend :return: list or results """ - return Task.get_result_group(group_id, failures) + if cached: + return result_group_cached(group_id, failures, wait, count) + start = time.time() + if count: + while True: + if count_group(group_id) == count or wait and (time.time() - start) * 1000 >= wait: + break + time.sleep(0.01) + while True: + r = Task.get_result_group(group_id, failures) + if r: + return r + if (time.time() - start) * 1000 >= wait: + break + time.sleep(0.01) -def fetch(task_id, wait=0): +def result_group_cached(group_id, failures=False, wait=0, count=None, broker=None): + """ + Return a list of results for a task group from the cache backend + """ + if not broker: + broker = get_broker() + start = time.time() + if count: + while True: + if count_group_cached(group_id) == count or wait and (time.time() - start) * 1000 >= wait: + break + time.sleep(0.01) + while True: + group_list = broker.cache.get('{}:{}:keys'.format(broker.list_key, group_id)) + if group_list: + result_list = [] + for task_key in group_list: + task = signing.SignedPackage.loads(broker.cache.get(task_key)) + if task['success'] or failures: + result_list.append(task['result']) + return result_list + if (time.time() - start) * 1000 >= wait: + break + time.sleep(0.01) + + +def fetch(task_id, wait=0, cached=Conf.CACHED): """ Return the processed task. @@ -123,9 +190,12 @@ def fetch(task_id, wait=0): :type task_id: str or uuid :param wait: the number of milliseconds to wait for a result :type wait: int + :param bool cached: run this against the cache backend :return: the full task object :rtype: Task """ + if cached: + return fetch_cached(task_id, wait) start = time.time() while True: t = Task.get_task(task_id) @@ -136,45 +206,169 @@ def fetch(task_id, wait=0): time.sleep(0.01) -def fetch_group(group_id, failures=True): +def fetch_cached(task_id, wait=0, broker=None): + """ + Return the processed task from the cache backend + """ + if not broker: + broker = get_broker() + start = time.time() + while True: + r = broker.cache.get('{}:{}'.format(broker.list_key, task_id)) + if r: + task = signing.SignedPackage.loads(r) + t = Task(id=task['id'], + name=task['name'], + func=task['func'], + hook=task.get('hook'), + args=task['args'], + kwargs=task['kwargs'], + started=task['started'], + stopped=task['stopped'], + result=task['result'], + success=task['success']) + return t + if (time.time() - start) * 1000 >= wait: + break + time.sleep(0.01) + + +def fetch_group(group_id, failures=True, wait=0, count=None, cached=Conf.CACHED): """ Return a list of Tasks for a task group. :param str group_id: the group id :param bool failures: set to False to exclude failures + :param bool cached: run this against the cache backend :return: list of Tasks """ - return Task.get_task_group(group_id, failures) + if cached: + return fetch_group_cached(group_id, failures, wait, count) + start = time.time() + if count: + while True: + if count_group(group_id) == count or wait and (time.time() - start) * 1000 >= wait: + break + time.sleep(0.01) + while True: + r = Task.get_task_group(group_id, failures) + if r: + return r + if (time.time() - start) * 1000 >= wait: + break + time.sleep(0.01) -def count_group(group_id, failures=False): +def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None): + """ + Return a list of Tasks for a task group in the cache backend + """ + if not broker: + broker = get_broker() + start = time.time() + if count: + while True: + if count_group_cached(group_id) == count or wait and (time.time() - start) * 1000 >= wait: + break + time.sleep(0.01) + while True: + group_list = broker.cache.get('{}:{}:keys'.format(broker.list_key, group_id)) + if group_list: + task_list = [] + for task_key in group_list: + task = signing.SignedPackage.loads(broker.cache.get(task_key)) + if task['success'] or failures: + t = Task(id=task['id'], + name=task['name'], + func=task['func'], + hook=task.get('hook'), + args=task['args'], + kwargs=task['kwargs'], + started=task['started'], + stopped=task['stopped'], + result=task['result'], + group=task.get('group'), + success=task['success']) + task_list.append(t) + return task_list + if (time.time() - start) * 1000 >= wait: + break + time.sleep(0.01) + + +def count_group(group_id, failures=False, cached=Conf.CACHED): """ Count the results in a group. :param str group_id: the group id :param bool failures: Returns failure count if True + :param bool cached: run this against the cache backend :return: the number of tasks/results in a group :rtype: int """ + if cached: + return count_group_cached(group_id, failures) return Task.get_group_count(group_id, failures) -def delete_group(group_id, tasks=False): +def count_group_cached(group_id, failures=False, broker=None): + """ + Count the results in a group in the cache backend + """ + if not broker: + broker = get_broker() + group_list = broker.cache.get('{}:{}:keys'.format(broker.list_key, group_id)) + if group_list: + if not failures: + return len(group_list) + failure_count = 0 + for task_key in group_list: + task = signing.SignedPackage.loads(broker.cache.get(task_key)) + if not task['success']: + failure_count += 1 + return failure_count + + +def delete_group(group_id, tasks=False, cached=Conf.CACHED): """ Delete a group. :param str group_id: the group id :param bool tasks: If set to True this will also delete the group tasks. Otherwise just the group label is removed. + :param bool cached: run this against the cache backend :return: """ + if cached: + return delete_group_cached(group_id) return Task.delete_group(group_id, tasks) +def delete_group_cached(group_id, broker=None): + """ + Delete a group from the cache backend + """ + if not broker: + broker = get_broker() + group_key = '{}:{}:keys'.format(broker.list_key, group_id) + group_list = broker.cache.get(group_key) + broker.cache.delete_many(group_list) + broker.cache.delete(group_key) + + +def delete_cached(task_id, broker=None): + """ + Delete a task from the cache backend + """ + if not broker: + broker = get_broker() + return broker.cache.delete('{}:{}'.format(broker.list_key, task_id)) + + def queue_size(broker=None): """ Returns the current queue size. - Note that this doesn't count any tasks currently being processed by workers. + Note that this doesn't count any tasks curren key = 'django_q:{}:results'.format(broker.list_key)tly being processed by workers. :param broker: optional broker :return: current queue size @@ -185,6 +379,26 @@ def queue_size(broker=None): return broker.queue_size() +def async_iter(func, args_iter, **kwargs): + iter_count = len(args_iter) + iter_group = uuid()[1] + # clean up the kwargs + options = kwargs.get('q_options', kwargs) + options.pop('hook', None) + options['broker'] = options.get('broker', get_broker()) + options['group'] = iter_group + options['iter_count'] = iter_count + options['cached'] = True + # save the original arguments + broker = options['broker'] + broker.cache.set('{}:{}:args'.format(broker.list_key, iter_group), signing.SignedPackage.dumps(args_iter)) + for args in args_iter: + if type(args) is not tuple: + args = (args,) + async(func, *args, **options) + return iter_group + + def _sync(pack): """Simulate a package travelling through the cluster.""" task_queue = Queue() diff --git a/django_q/tests/test_cached.py b/django_q/tests/test_cached.py new file mode 100644 index 0000000..ee05e36 --- /dev/null +++ b/django_q/tests/test_cached.py @@ -0,0 +1,87 @@ +from multiprocessing import Event + +import pytest + +from django_q.cluster import Sentinel +from django_q.conf import Conf +from django_q.tasks import async, result, fetch, count_group, result_group, fetch_group, delete_group, delete_cached, \ + async_iter +from django_q.brokers import get_broker + + +@pytest.fixture +def broker(): + Conf.DISQUE_NODES = None + Conf.IRON_MQ = None + Conf.SQS = None + Conf.ORM = None + Conf.MONGO = None + Conf.DJANGO_REDIS = 'default' + return get_broker() + + +@pytest.mark.django_db +def test_cached(broker): + broker.purge_queue() + broker.cache.clear() + group = 'cache_test' + # queue the tests + task_id = async('math.copysign', 1, -1, cached=True, broker=broker) + async('math.copysign', 1, -1, cached=True, broker=broker, group=group) + async('math.copysign', 1, -1, cached=True, broker=broker, group=group) + async('math.copysign', 1, -1, cached=True, broker=broker, group=group) + async('math.copysign', 1, -1, cached=True, broker=broker, group=group) + async('math.copysign', 1, -1, cached=True, broker=broker, group=group) + async('math.popysign', 1, -1, cached=True, broker=broker, group=group) + # test wait on cache + # test wait timeout + assert result(task_id, wait=10, cached=True) is None + assert fetch(task_id, wait=10, cached=True) is None + assert result_group(group, wait=10, cached=True) is None + assert result_group(group, count=2, wait=10, cached=True) is None + assert fetch_group(group, wait=10, cached=True) is None + assert fetch_group(group, count=2, wait=10, cached=True) is None + # run a single cluster + start_event = Event() + stop_event = Event() + stop_event.set() + Sentinel(stop_event, start_event, broker=broker) + # assert results + assert result(task_id, wait=500, cached=True) == -1 + assert fetch(task_id, wait=500, cached=True).result == -1 + # make sure it's not in the db backend + assert fetch(task_id) is None + # assert group + assert count_group(group, cached=True) == 6 + assert count_group(group, cached=True, failures=True) == 1 + assert result_group(group, cached=True) == [-1, -1, -1, -1, -1] + assert len(result_group(group, cached=True, failures=True)) == 6 + assert len(fetch_group(group, cached=True)) == 6 + assert len(fetch_group(group, cached=True, failures=False)) == 5 + delete_group(group, cached=True) + assert count_group(group, cached=True) is None + delete_cached(task_id) + assert result(task_id, cached=True) is None + assert fetch(task_id, cached=True) is None + broker.cache.clear() + + +@pytest.mark.django_db +def test_iter(broker): + broker.purge_queue() + broker.cache.clear() + it = [i for i in range(10)] + it2 = [(1, -1), (2, -1), (3, -4), (5, 6)] + it3 = (1, 2, 3, 4, 5) + t = async_iter('math.floor', it, sync=True) + t2 = async_iter('math.copysign', it2, sync=True) + t3 = async_iter('math.floor', it3, sync=True) + t4 = async_iter('math.floor', (1,), sync=True) + result_t = result(t) + assert result_t is not None + task_t = fetch(t) + assert task_t. __unicode__ is not None + assert task_t.result == result_t + assert result(t2) is not None + assert result(t3) is not None + assert result(t4)[0] == 1 diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 37dc79a..5b1350e 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -160,6 +160,13 @@ def test_async(broker, admin_user): assert broker.queue_size() == 0 assert task_queue.qsize() == task_count task_queue.put('STOP') + # test wait timeout + assert result(j, wait=10) is None + assert fetch(j, wait=10) is None + assert result_group('test_j', wait=10) is None + assert result_group('test_j', count=2, wait=10) is None + assert fetch_group('test_j', wait=10) is None + assert fetch_group('test_j', count=2, wait=10) is None # let a worker handle them result_queue = Queue() worker(task_queue, result_queue, Value('f', -1)) diff --git a/docs/configure.rst b/docs/configure.rst index 1795670..9f07d4b 100644 --- a/docs/configure.rst +++ b/docs/configure.rst @@ -308,6 +308,13 @@ cache For some brokers, you will need to set up the Django `cache framework `__ to gather statistics for the monitor. You can indicate which cache to use by setting this value. Defaults to ``default``. +.. _cached: + +cached +~~~~~~ +Switches all task and result functions from using the database backend to the cache backend. This is the same as setting the keyword ``cached=True`` on all task functions. +Instead of a bool this can also be set to the number of seconds you want the cache to retain results. e.g. ``cached=60`` + scheduler ~~~~~~~~~ You can disable the scheduler by setting this option to ``False``. This will reduce a little overhead if you're not using schedules, but is most useful if you want to temporarily disable all schedules. diff --git a/docs/examples.rst b/docs/examples.rst index 3ab2219..15ce2a2 100644 --- a/docs/examples.rst +++ b/docs/examples.rst @@ -253,8 +253,7 @@ Adapted from `Sebastian Raschka's blog `__. diff --git a/docs/index.rst b/docs/index.rst index b46ba00..761b8a6 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -15,7 +15,7 @@ Features - Asynchronous tasks - Scheduled and repeated tasks - Encrypted and compressed packages -- Failure and success database +- Failure and success database or cache - Result hooks and groups - Django Admin integration - PaaS compatible with multiple instances diff --git a/docs/tasks.rst b/docs/tasks.rst index 5da69c1..f1a0fb1 100644 --- a/docs/tasks.rst +++ b/docs/tasks.rst @@ -59,6 +59,12 @@ sync Simulates a task execution synchronously. Useful for testing. Can also be forced globally via the :ref:`sync` configuration option. +cached +"""""" +Redirects the result to the cache backend instead of the database if set to ``True`` or to an integer indicating the cache timeout in seconds. +e.g. ``cached=60``. Especially useful with large and group operations where you don't need the all results in your +database and want to take advantage of the speed of your cache backend. + broker """""" A broker instance, in case you want to control your own connections. @@ -83,6 +89,28 @@ Please not that this will override any other option keywords. For tasks to be processed you will need to have a worker cluster running in the background using ``python manage.py qcluster`` or you need to configure Django Q to run in synchronous mode for testing using the :ref:`sync` option. + + +Async Iterable +-------------- +If you have an iterable object with arguments for a function, you can use :func:`async_iter` to async them with a single command:: + + # Async Iterable example + from django_q.tasks import async_iter, result + + # set up a list of arguments for math.floor + iter = [i for i in range(100)] + + # async iter them + id=async_iter('math.floor',iter) + + # wait for the collated result for 1 second + result_list = result(id, wait=1000) + +This will individually queue 100 tasks to the worker cluster, which will save their results in the cache backend for speed. +Once all the 100 results are in the cache, they are collated into a list and saved as a single result in the database. The cache results are then cleared. +Needs the Django cache framework. + .. _groups: Groups @@ -97,14 +125,16 @@ You can group together results by passing :func:`async` the optional ``group`` k for i in range(4): async('math.modf', i, group='modf') - # after the tasks have finished you can get the group results - result = result_group('modf') + # wait until the group has 4 results + result = result_group('modf', count=4) print(result) .. code-block:: python [(0.0, 0.0), (0.0, 1.0), (0.0, 2.0), (0.0, 3.0)] +Note that the same can be achieved much faster with :func:`async_iter` + Take care to not limit your results database too much and call :func:`delete_group` before each run, unless you want your results to keep adding up. Instead of :func:`result_group` you can also use :func:`fetch_group` to return a queryset of :class:`Task` objects.: @@ -149,6 +179,54 @@ You can also access group functions from a task result instance: task.group_delete() print('Deleted group {}'.format(task.group)) +Cached operations +----------------- +You can run your tasks results against the Django cache backend instead of the database backend by either using the global :ref:`cached` setting or by supplying the ``cached`` keyword to individual functions. +This can be useful if you are not interested in persistent results or if you run large group tasks where you only want the final result. +By using a cache backend like Redis or Memcached you can speed up access to your task results significantly compared to a relational database. + +When you set ``cached=True``, results will be saved permanently in the cache and you will have to rely on your backend's cleanup strategies (like LRU) to +manage stale results. +You can also opt to set a manual timeout on the results, by setting ``cached=60``. Meaning the result will be evicted from the cache after 60 seconds. +This works both globally or on individual async executions.:: + + # simple cached example + from django_q.tasks import async, result + + # cache the result for 10 seconds + id = async('math.floor', 100, cached=10) + + # wait max 50ms for the result to appear in the cache + result(id, wait=50, cached=True) + + # o fetch the task object + task = fetch(id, cache=True) + + # and then save it to the database + task.save() + +This also works for group actions:: + + # cached group example + from django_q.tasks import async, result_group + from django_q.brokers import get_broker + + # set up a broker instance for better performance + broker = get_broker() + + # async a hundred functions under a group label + for i in range(100): + async('math.frexp', + i, + group='frexp', + cached=True, + broker=broker) + + # wait max 50ms for one hundred results to return + result_group('frexp', wait=50, count=100, cached=True) + +Note that exact same result can be achieved by using the more convenient :func:`async_iter` in this case, but without hook support. + Synchronous testing ------------------- @@ -199,7 +277,7 @@ Reference --------- .. py:function:: async(func, *args, hook=None, group=None, timeout=None,\ - save=None, sync=False, broker=None, q_options=None, **kwargs) + save=None, sync=False, cached=False, broker=None, q_options=None, **kwargs) Puts a task in the cluster queue @@ -210,26 +288,29 @@ Reference :param int timeout: Overrides global cluster :ref:`timeout`. :param bool save: Overrides global save setting for this task. :param bool sync: If set to True, async will simulate a task execution + :param cached: Output the result to the cache backend. Bool or timeout in seconds :param broker: Optional broker connection from :func:`brokers.get_broker` :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 -.. py:function:: result(task_id, wait=0) +.. py:function:: result(task_id, wait=0, cached=False) Gets the result of a previously executed task :param str task_id: the uuid or name of the task :param int wait: optional milliseconds to wait for a result + :param bool cached: run this against the cache backend. :returns: The result of the executed task -.. py:function:: fetch(task_id, wait=0) +.. py:function:: fetch(task_id, wait=0, cached=False) Returns a previously executed task :param str name: the uuid or name of the task :param int wait: optional milliseconds to wait for a result + :param bool cached: run this against the cache backend. :returns: A task object :rtype: Task @@ -237,6 +318,16 @@ Reference Renamed from get_task +.. py:function:: async_iter(func, args_iter,**kwargs) + + Runs iterable arguments against the cache backend and returns a single collated result + + :param object func: The task function to execute + :param args: An iterable containing arguments for the task function + :param dict kwargs: Keyword arguments for the task function. Ignores ``cached`` and ``hook``. + :returns: The uuid of the task + :rtype: str + .. py:function:: queue_size() Returns the size of the broker queue. @@ -245,42 +336,58 @@ Reference :returns: The amount of task packages in the broker :rtype: int -.. py:function:: result_group(group_id, failures=False) +.. py:function:: result_group(group_id, failures=False, wait=0, count=None, cached=False) 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 int wait: optional milliseconds to wait for a result or count + :param int count: block until there are this many results in the group + :param bool cached: run this against the cache backend :returns: a list of results :rtype: list -.. py:function:: fetch_group(group_id, failures=True) +.. py:function:: fetch_group(group_id, failures=True, wait=0, count=None, cached=False) 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 - :returns: a list of Tasks + :param int wait: optional milliseconds to wait for a task or count + :param int count: block until there are this many tasks in the group + :param bool cached: run this against the cache backend. + :returns: a list of :class:`Task` :rtype: list -.. py:function:: count_group(group_id, failures=False) +.. py:function:: count_group(group_id, failures=False, cached=False) 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 cached: run this against the cache backend. :returns: the number of tasks or failures in a group :rtype: int -.. py:function:: delete_group(group_id, tasks=False) +.. py:function:: delete_group(group_id, tasks=False, cached=False) 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 cached: run this against the cache backend. :returns: the numbers of tasks affected :rtype: int +.. py:function:: delete_cached(task_id, broker=None) + + Deletes a task from the cache backend + + :param task_id: the uuid of the task + :param broker: an optional broker instance + + .. py:class:: Task Database model describing an executed task