diff --git a/tasktiger/task.py b/tasktiger/task.py index 2343ba60..d5b5bb5b 100644 --- a/tasktiger/task.py +++ b/tasktiger/task.py @@ -513,15 +513,19 @@ def tasks_from_queue( tiger: TaskTiger instance. queue: Name of the queue. state: State of the task (QUEUED, ACTIVE, SCHEDULED, ERROR). + skip: Number of latest tasks to pass over. limit: Maximum number of tasks to return. load_executions: Maximum number of executions to load for each task (starting from the latest). - include_not_found: Whether to include tasks that cannot be loaded. + include_not_found: Whether to include tasks that have no payload. Returns: Tuple with the following information: - * total items in the queue - * tasks from the given queue in the given state, latest first. + * total items in the queue, including tasks without a payload + * the latest `limit` tasks after the latest `skip` tasks, in + ascending order of score. If tasks without a payload are left + out, this list can be shorter than `limit` before the end of the + queue. """ key = tiger._key(state, queue) @@ -549,8 +553,9 @@ def tasks_from_queue( range(len(items)), results[0], results[1:], tss ): if serialized_data is None: - if include_not_found: - data = {"id": items[idx][0]} + if not include_not_found: + continue + data = {"id": items[idx][0]} else: data = json.loads(serialized_data) @@ -572,8 +577,9 @@ def tasks_from_queue( ) for idx, serialized_data, ts in zip(range(len(items)), result, tss): if serialized_data is None: - if include_not_found: - data = {"id": items[idx][0]} + if not include_not_found: + continue + data = {"id": items[idx][0]} else: data = json.loads(serialized_data) diff --git a/tests/test_base.py b/tests/test_base.py index df05ea1c..26c7a79b 100644 --- a/tests/test_base.py +++ b/tests/test_base.py @@ -1072,6 +1072,35 @@ def test_tasks_from_queue_with_executions(self): assert n == 1 assert len(tasks[0].executions) == 2 + @pytest.mark.parametrize("load_executions", [0, 1]) + @pytest.mark.parametrize("missing_index", [0, 1, 2]) + def test_tasks_from_queue_task_not_found(self, load_executions, missing_index): + tasks = [Task(self.tiger, simple_task) for _ in range(3)] + # Distinct scheduled times give a fixed order in the zset. + for minutes, task in enumerate(tasks, start=1): + task.delay(when=datetime.timedelta(minutes=minutes)) + missing_task = tasks[missing_index] + self.conn.delete(self.tiger._key("task", missing_task.id)) + + n, found = Task.tasks_from_queue( + self.tiger, "default", "scheduled", load_executions=load_executions + ) + assert n == 3 + assert [task.id for task in found] == [ + task.id for task in tasks if task is not missing_task + ] + + n, found = Task.tasks_from_queue( + self.tiger, + "default", + "scheduled", + load_executions=load_executions, + include_not_found=True, + ) + assert n == 3 + assert [task.id for task in found] == [task.id for task in tasks] + assert found[missing_index].data == {"id": missing_task.id} + def test_eager(self): self.tiger.config["ALWAYS_EAGER"] = True