Method tasks_from_queue

tasktiger/task.py:499–583

Return tasks from a queue. Args: tiger: TaskTiger instance. queue: Name of the queue. state: State of the task (QUEUED, ACTIVE, SCHEDULED, ERROR). limit: Maximum number of tasks to return. load_executions: Maximum number o

(
        cls,
        tiger: "TaskTiger",
        queue: str,
        state: str,
        skip: int = 0,
        limit: int = 1000,
        load_executions: int = 0,
        include_not_found: bool = False,
    )

    @classmethod
    def tasks_from_queue(
        cls,
        tiger: "TaskTiger",
        queue: str,
        state: str,
        skip: int = 0,
        limit: int = 1000,
        load_executions: int = 0,
        include_not_found: bool = False,
    ) -> Tuple[int, List["Task"]]:
        """
        Return tasks from a queue.

        Args:
            tiger: TaskTiger instance.
            queue: Name of the queue.
            state: State of the task (QUEUED, ACTIVE, SCHEDULED, ERROR).
            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.

        Returns:
            Tuple with the following information:
            * total items in the queue
            * tasks from the given queue in the given state, latest first.
        """

        key = tiger._key(state, queue)
        pipeline = tiger.connection.pipeline()
        pipeline.zcard(key)
        pipeline.zrange(key, -limit - skip, -1 - skip, withscores=True)
        n, items = pipeline.execute()

        tasks = []

        if items:
            tss = [datetime.datetime.utcfromtimestamp(item[1]) for item in items]
            if load_executions:
                pipeline = tiger.connection.pipeline()
                pipeline.mget([tiger._key("task", item[0]) for item in items])
                for item in items:
                    pipeline.lrange(
                        tiger._key("task", item[0], "executions"),
                        -load_executions,
                        -1,
                    )
                results = pipeline.execute()

                for idx, serialized_data, serialized_executions, ts in zip(
                    range(len(items)), results[0], results[1:], tss
                ):
                    if serialized_data is None:
                        if include_not_found:
                            data = {"id": items[idx][0]}
                    else:
                        data = json.loads(serialized_data)