def _summary_format()

in luigi/execution_summary.py [0:0]

• 44 lines of code
• 13 McCabe index (conditional complexity)

def _summary_format(set_tasks, worker):
    group_tasks = {}
    for status, task_dict in set_tasks.items():
        group_tasks[status] = _group_tasks_by_name_and_status(task_dict)
    comments = _get_comments(group_tasks)
    num_all_tasks = sum([len(set_tasks["already_done"]),
                         len(set_tasks["completed"]), len(set_tasks["failed"]),
                         len(set_tasks["scheduling_error"]),
                         len(set_tasks["still_pending_ext"]),
                         len(set_tasks["still_pending_not_ext"])])
    str_output = ''
    str_output += 'Scheduled {0} tasks of which:\n'.format(num_all_tasks)
    for status in _ORDERED_STATUSES:
        if status not in comments:
            continue
        str_output += '{0}'.format(comments[status])
        if status != 'still_pending':
            str_output += '{0}\n'.format(_get_str(group_tasks[status], status in _PENDING_SUB_STATUSES))
    ext_workers = _get_external_workers(worker)
    group_tasks_ext_workers = {}
    for ext_worker, task_dict in ext_workers.items():
        group_tasks_ext_workers[ext_worker] = _group_tasks_by_name_and_status(task_dict)
    if len(ext_workers) > 0:
        str_output += "\nThe other workers were:\n"
        count = 0
        for ext_worker, task_dict in ext_workers.items():
            if count > 3 and count < len(ext_workers) - 1:
                str_output += "    and {0} other workers".format(len(ext_workers) - count)
                break
            str_output += "    - {0} ran {1} tasks\n".format(ext_worker, len(task_dict))
            count += 1
        str_output += '\n'
    if num_all_tasks == sum([len(set_tasks["already_done"]),
                             len(set_tasks["scheduling_error"]),
                             len(set_tasks["still_pending_ext"]),
                             len(set_tasks["still_pending_not_ext"])]):
        if len(ext_workers) == 0:
            str_output += '\n'
        str_output += 'Did not run any tasks'
    one_line_summary = _create_one_line_summary(_tasks_status(set_tasks))
    str_output += "\n{0}".format(one_line_summary)
    if num_all_tasks == 0:
        str_output = 'Did not schedule any tasks'
    return str_output