diff --git a/CHANGELOG b/CHANGELOG index c240c7a157..98c7d706ee 100644 --- a/CHANGELOG +++ b/CHANGELOG @@ -1,3 +1,4 @@ +* Add batch option to push queue tasks * Support ref value for reference field * Enforce ref model in XML data * Do not resolve entities by default with lxml (issue11219) diff --git a/doc/topics/configuration.rst b/doc/topics/configuration.rst index fd1cf44a5d..6469ff53e6 100644 --- a/doc/topics/configuration.rst +++ b/doc/topics/configuration.rst @@ -311,6 +311,13 @@ The number of days after which processed tasks are removed. Default: ``30`` +batch_size +~~~~~~~~~~ + +The default number of the instances to process in a batch. + +Default: ``20`` + error ----- diff --git a/doc/topics/task_queue.rst b/doc/topics/task_queue.rst index 8d78f3d416..eb06c95cc6 100644 --- a/doc/topics/task_queue.rst +++ b/doc/topics/task_queue.rst @@ -34,6 +34,11 @@ parameters for the queue: define when the task should be finished. Default value is ``None`` which means as soon as possible. + - ``queue_batch``: An ``integer`` to divide the instances by batch of this + size. If the value is ``true`` then the size is the value defined by the + configuration ``queue`` of ``batch_size``. + Default is ``None`` which means no division. + .. warning:: There is no access right verification during the execution of the task. diff --git a/trytond/ir/queue.py b/trytond/ir/queue.py index 60b19c0982..6b963c00a3 100644 --- a/trytond/ir/queue.py +++ b/trytond/ir/queue.py @@ -14,6 +14,7 @@ from trytond.transaction import Transaction has_worker = config.getboolean('queue', 'worker', default=False) clean_days = config.getint('queue', 'clean_days', default=30) +batch_size = config.getint('queue', 'batch_size', default=20) class Queue(ModelSQL): @@ -221,6 +222,7 @@ class _Method(object): if scheduled_at is not None: scheduled_at = now + scheduled_at expected_at = context.pop('queue_expected_at', None) + queue_batch = context.pop('queue_batch', None) context.pop('_check_access', None) context.pop('language', None) if expected_at is not None: @@ -229,15 +231,32 @@ class _Method(object): instances = list(map(int, instances)) except TypeError: instances = int(instances) - data = { - 'model': self.__model.__name__, - 'method': self.__name, - 'user': transaction.user, - 'context': context, - 'instances': instances, - 'args': args, - 'kwargs': kwargs, - } - return self.__queue.push( - name, data, - scheduled_at=scheduled_at, expected_at=expected_at) + + def _push(instances): + data = { + 'model': self.__model.__name__, + 'method': self.__name, + 'user': transaction.user, + 'context': context, + 'instances': instances, + 'args': args, + 'kwargs': kwargs, + } + return self.__queue.push( + name, data, + scheduled_at=scheduled_at, expected_at=expected_at) + + if isinstance(instances, list): + if has_worker and queue_batch: + if isinstance(queue_batch, int): + count = queue_batch + else: + count = batch_size + else: + count = len(instances) + task_ids = [] + for sub_instances in grouped_slice(instances, count=count): + task_ids.append(_push(list(sub_instances))) + return task_ids + else: + return _push(instances)