[IMP] queue_job: Add split method

queue_job/delay.py

@@ -534,6 +534,50 @@ def delay(self):
         """Delay the whole graph"""
         self._graph.delay()
 
+    def split(self, size):
+        """Split the Delayable into a DelayableGroup containing batches
+        of size `size`
+        """
+        if not self._job_method:
+            raise ValueError("No method set on the Delayable")
+
+        total_records = len(self.recordset)
+
+        delayables = []
+        for index in range(0, total_records, size):
+            recordset = self.recordset[index : index + size]
+            delayable = Delayable(
+                recordset,
+                priority=self.priority,
+                eta=self.eta,
+                max_retries=self.max_retries,
+                description=self.description,
+                channel=self.channel,
+                identity_key=self.identity_key,
+            )
+            # Update the __self__
+            delayable._job_method = getattr(recordset, self._job_method.__name__)
+            delayable._job_args = self._job_args
+            delayable._job_kwargs = self._job_kwargs
+
+            delayables.append(delayable)
+
+        description = self.description or (
+            self._job_method.__doc__.splitlines()[0].strip()
+            if self._job_method.__doc__
+            else "{}.{}".format(self.recordset._name, self._job_method.__name__)
+        )
+        for index, delayable in enumerate(delayables):
+            delayable.set(
+                description="%s (split %s/%s)"
+                % (description, index + 1, len(delayables))
+            )
+
+        # Prevent warning on deletion
+        self._generated_job = True
+
+        return DelayableGroup(*delayables)
+
     def _build_job(self):
         if self._generated_job:
             return self._generated_job

queue_job/tests/test_delayable.py

@@ -271,3 +271,89 @@ def test_graph_topological_sort(self):
         ]
 
         self.assertIn(list(graph.topological_sort()), valid_solutions)
+
+
+class TestDelayableSplit(unittest.TestCase):
+    def setUp(self):
+        super().setUp()
+
+        class FakeRecordSet(list):
+            def __init__(self, *args, **kwargs):
+                super().__init__(*args, **kwargs)
+                self._name = "recordset"
+
+            def __getitem__(self, key):
+                if isinstance(key, slice):
+                    return FakeRecordSet(super().__getitem__(key))
+                return super().__getitem__(key)
+
+            def method(self, arg, kwarg=None):
+                """Method to be called"""
+                return arg, kwarg
+
+        self.FakeRecordSet = FakeRecordSet
+
+    def test_delayable_split_no_method_call_beforehand(self):
+        dl = Delayable(self.FakeRecordSet(range(20)))
+        with self.assertRaises(ValueError):
+            dl.split(3)
+
+    def test_delayable_split_10_3(self):
+        dl = Delayable(self.FakeRecordSet(range(10)))
+        dl.method("arg", kwarg="kwarg")
+        group = dl.split(3)
+        self.assertEqual(len(group._delayables), 4)
+        delayables = sorted(list(group._delayables), key=lambda x: x.description)
+        self.assertEqual(delayables[0].recordset, self.FakeRecordSet([0, 1, 2]))
+        self.assertEqual(delayables[1].recordset, self.FakeRecordSet([3, 4, 5]))
+        self.assertEqual(delayables[2].recordset, self.FakeRecordSet([6, 7, 8]))
+        self.assertEqual(delayables[3].recordset, self.FakeRecordSet([9]))
+        self.assertEqual(delayables[0].description, "Method to be called (split 1/4)")
+        self.assertEqual(delayables[1].description, "Method to be called (split 2/4)")
+        self.assertEqual(delayables[2].description, "Method to be called (split 3/4)")
+        self.assertEqual(delayables[3].description, "Method to be called (split 4/4)")
+        self.assertNotEqual(delayables[0]._job_method, dl._job_method)
+        self.assertNotEqual(delayables[1]._job_method, dl._job_method)
+        self.assertNotEqual(delayables[2]._job_method, dl._job_method)
+        self.assertNotEqual(delayables[3]._job_method, dl._job_method)
+        self.assertEqual(delayables[0]._job_method.__name__, dl._job_method.__name__)
+        self.assertEqual(delayables[1]._job_method.__name__, dl._job_method.__name__)
+        self.assertEqual(delayables[2]._job_method.__name__, dl._job_method.__name__)
+        self.assertEqual(delayables[3]._job_method.__name__, dl._job_method.__name__)
+        self.assertEqual(delayables[0]._job_args, ("arg",))
+        self.assertEqual(delayables[1]._job_args, ("arg",))
+        self.assertEqual(delayables[2]._job_args, ("arg",))
+        self.assertEqual(delayables[3]._job_args, ("arg",))
+        self.assertEqual(delayables[0]._job_kwargs, {"kwarg": "kwarg"})
+        self.assertEqual(delayables[1]._job_kwargs, {"kwarg": "kwarg"})
+        self.assertEqual(delayables[2]._job_kwargs, {"kwarg": "kwarg"})
+        self.assertEqual(delayables[3]._job_kwargs, {"kwarg": "kwarg"})
+
+    def test_delayable_split_10_5(self):
+        dl = Delayable(self.FakeRecordSet(range(10)))
+        dl.method("arg", kwarg="kwarg")
+        group = dl.split(5)
+        self.assertEqual(len(group._delayables), 2)
+        delayables = sorted(list(group._delayables), key=lambda x: x.description)
+        self.assertEqual(delayables[0].recordset, self.FakeRecordSet([0, 1, 2, 3, 4]))
+        self.assertEqual(delayables[1].recordset, self.FakeRecordSet([5, 6, 7, 8, 9]))
+        self.assertEqual(delayables[0].description, "Method to be called (split 1/2)")
+        self.assertEqual(delayables[1].description, "Method to be called (split 2/2)")
+
+    def test_delayable_split_10_10(self):
+        dl = Delayable(self.FakeRecordSet(range(10)))
+        dl.method("arg", kwarg="kwarg")
+        group = dl.split(10)
+        self.assertEqual(len(group._delayables), 1)
+        delayables = sorted(list(group._delayables), key=lambda x: x.description)
+        self.assertEqual(delayables[0].recordset, self.FakeRecordSet(range(10)))
+        self.assertEqual(delayables[0].description, "Method to be called (split 1/1)")
+
+    def test_delayable_split_10_20(self):
+        dl = Delayable(self.FakeRecordSet(range(10)))
+        dl.method("arg", kwarg="kwarg")
+        group = dl.split(20)
+        self.assertEqual(len(group._delayables), 1)
+        delayables = sorted(list(group._delayables), key=lambda x: x.description)
+        self.assertEqual(delayables[0].recordset, self.FakeRecordSet(range(10)))
+        self.assertEqual(delayables[0].description, "Method to be called (split 1/1)")