state.py 文件源码

def _assemble_tasks(self, task_list=None):
        """Attach all the appropriate result messages to the tasks in the list.

        :param task_list: a list of objects.Task instances to attach result messages to
        """
        if task_list is None:
            return None

        conn = self.db_engine.connect()
        query = sql.select([
            self.result_message_tbl
        ]).where(self.result_message_tbl.c.task_id == sql.bindparam(
            'task_id')).order_by(self.result_message_tbl.c.sequence.asc())
        query.compile(self.db_engine)

        for t in task_list:
            rs = conn.execute(query, task_id=t.task_id.bytes)
            error_count = 0
            for r in rs:
                msg = objects.TaskStatusMessage.from_db(dict(r))
                if msg.error:
                    error_count = error_count + 1
                t.result.message_list.append(msg)
            t.result.error_count = error_count

        conn.close()