Repository: spotify/luigi
File: luigi/execution_summary.py (Python)

Code before:
    SCHEDULING_FAILED              There were tasks whose scheduling failed
    NOT_RUN                        There were tasks that were not granted run permission by the scheduler
    MISSING_EXT                    There were missing external dependencies
    =============================  ==========================================================

    """
    SUCCESS = (":)", "there were no failed tasks or missing dependencies")
    SUCCESS_WITH_RETRY = (":)", "there were failed tasks but they all succeeded in a retry")
    FAILED = (":(", "there were failed tasks")
    FAILED_AND_SCHEDULING_FAILED = (":(", "there were failed tasks and tasks whose scheduling failed")
    SCHEDULING_FAILED = (":(", "there were tasks whose scheduling failed")
    NOT_RUN = (":|", "there were tasks that were not granted run permission by the scheduler")
    MISSING_EXT = (":|", "there were missing external dependencies")


class LuigiRunResult:
    """
    Result of the execution (build/run) will be of type LuigiRunResult instead of
    the regular Boolean response if the keyword argument ``detailed_summary=True`` is passed to
    build/run. A response of type **LuigiRunResult** has the following attributes:

    Attributes:
        - summary_text_one_line (str): One line summary of the progress.
        - summary_text (str): Detailed summary of the progress.
        - status (LuigiStatusCode): Luigi Status Code. See \
        `LuigiStatusCode <https://luigi.readthedocs.io/en/latest/api/luigi.execution_summary.html#luigi.execution_summary.LuigiStatusCode>`_ \
        for what these codes mean.
        - worker (luigi.worker.worker): Worker object.
        - execution_succeeded (bool): Boolean which is *True* if finally, there were no failed tasks.
        - scheduling_succeeded (bool): Boolean which is *True* if all the tasks were scheduled without errors.

    """
    def __init__(self, worker, worker_add_run_status=True):
        self.worker = worker
        self.summary_text = summary(worker)
        self.status = _tasks_status(_summary_dict(worker))
        self.summary_text_one_line = _progress_summary(self.status)
        self.scheduling_succeeded = worker_add_run_status
        self.execution_succeeded = self.status in (LuigiStatusCode.SUCCESS, LuigiStatusCode.SUCCESS_WITH_RETRY)

    # This function makes this class subscriptable (self.worker is used at a few places)
    def __getitem__(self, key):
        return getattr(self, key, None)

    def __str__(self):
        return "STATUS: {0}".format(self.status, self.summary_text_one_line)

    def __repr__(self):
        return self.__str__()



Review comment (suggestion):
I think there's a better way to link to classes. I think you will find it for example in the parameter.py file.

Code after:
    FAILED_AND_SCHEDULING_FAILED   There were failed tasks and tasks whose scheduling failed
    SCHEDULING_FAILED              There were tasks whose scheduling failed
    NOT_RUN                        There were tasks that were not granted run permission by the scheduler
    MISSING_EXT                    There were missing external dependencies
    =============================  ==========================================================

    """
    SUCCESS = (":)", "there were no failed tasks or missing dependencies")
    SUCCESS_WITH_RETRY = (":)", "there were failed tasks but they all succeeded in a retry")
    FAILED = (":(", "there were failed tasks")
    FAILED_AND_SCHEDULING_FAILED = (":(", "there were failed tasks and tasks whose scheduling failed")
    SCHEDULING_FAILED = (":(", "there were tasks whose scheduling failed")
    NOT_RUN = (":|", "there were tasks that were not granted run permission by the scheduler")
    MISSING_EXT = (":|", "there were missing external dependencies")


class LuigiRunResult(object):
    """
    The result of a call to build/run when passing the detailed_summary=True argument.

    Attributes:
        - one_line_summary (str): One line summary of the progress.
        - summary_text (str): Detailed summary of the progress.
        - status (LuigiStatusCode): Luigi Status Code. See :class:`~luigi.execution_summary.LuigiStatusCode` for what these codes mean.
        - worker (luigi.worker.worker): Worker object. See :class:`~luigi.worker.worker`.
        - scheduling_succeeded (bool): Boolean which is *True* if all the tasks were scheduled without errors.

    """
    def __init__(self, worker, worker_add_run_status=True):
        self.worker = worker
        summary_dict = _summary_dict(worker)
        self.summary_text = _summary_wrap(_summary_format(summary_dict, worker))
        self.status = _tasks_status(summary_dict)
        self.one_line_summary = _create_one_line_summary(self.status)
        self.scheduling_succeeded = worker_add_run_status

    def __str__(self):
        return "LuigiRunResult with status {0}".format(self.status)

    def __repr__(self):
        return "LuigiRunResult(status={0!r},worker={1!r},scheduling_succeeded={2!r})".format(self.status, self.worker, self.scheduling_succeeded)


def _partition_tasks(worker):
    """
    Takes a worker and sorts out tasks based on their status.
    Still_pending_not_ext is only used to get upstream_failure, upstream_missing_dependency and run_by_other_worker
    """
    task_history = worker._add_task_history
    pending_tasks = {task for(task, status, ext) in task_history if status == 'PENDING'}
    set_tasks = {}