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

Code before:
                    elif upstream_status_table[dep_id] == '' and dep.deps:
                        # This is the postorder update step when we set the
                        # status based on the previously calculated child elements
                        status = max((upstream_status_table.get(a_task_id, '')
                                      for a_task_id in dep.deps),
                                     key=UPSTREAM_SEVERITY_KEY)
                        upstream_status_table[dep_id] = status
            return upstream_status_table[dep_id]

    def _serialize_task(self, task_id, include_deps=True, deps=None):
        task = self._state.get_task(task_id)
        ret = {
            'display_name': task.pretty_id,
            'status': task.status,
            'workers': list(task.workers),
            'worker_running': task.worker_running,
            'time_running': getattr(task, "time_running", None),
            'start_time': task.time,
            'last_updated': getattr(task, "updated", task.time),
            'params': task.params,
            'name': task.family,
            'priority': task.priority,
            'resources': task.resources,
            'tracking_url': getattr(task, "tracking_url", None),
        }
        if task.status == DISABLED:
            ret['re_enable_able'] = task.scheduler_disable_time is not None
        if include_deps:
            ret['deps'] = list(task.deps if deps is None else deps)
        return ret

    def graph(self, **kwargs):
        self.prune()
        serialized = {}
        seen = set()
        for task in self._state.get_active_tasks():
            serialized.update(self._traverse_graph(task.id, seen))
        return serialized

    def _filter_done(self, task_ids):
        for task_id in task_ids:
            task = self._state.get_task(task_id)
            if task is None or task.status != DONE:
                yield task_id

    def _traverse_graph(self, root_task_id, seen=None, dep_func=None, include_done=True):
        """ Returns the dependency graph rooted at task_id

        This does a breadth-first traversal to find the nodes closest to the
        root before hitting the scheduler.max_graph_nodes limit.


Review comment (suggestion):
This line is causing a stack-trace like this:

```
Traceback (most recent call last):
  File "/zserver/Python-3.4.3/lib/python3.4/site-packages/tornado/web.py", line 1443, in _execute
    result = method(*self.path_args, **self.path_kwargs)
  File "/zserver/apps/luigi/code-for-luigid/luigi/server.py", line 101, in get
    result = getattr(self._scheduler, method)(**arguments)
  File "/zserver/apps/luigi/code-for-luigid/luigi/scheduler.py", line 986, in task_list
    serialized = self._serialize_task(task.id, False)
  File "/zserver/apps/luigi/code-for-luigid/luigi/scheduler.py", line 868, in _serialize_task
    'status_message': task.status_message
```

For anyone using an old pickled state-file.


Code after:
                            upstream_status_table[dep_id] = dep_status
                    elif upstream_status_table[dep_id] == '' and dep.deps:
                        # This is the postorder update step when we set the
                        # status based on the previously calculated child elements
                        status = max((upstream_status_table.get(a_task_id, '')
                                      for a_task_id in dep.deps),
                                     key=UPSTREAM_SEVERITY_KEY)
                        upstream_status_table[dep_id] = status
            return upstream_status_table[dep_id]

    def _serialize_task(self, task_id, include_deps=True, deps=None):
        task = self._state.get_task(task_id)
        ret = {
            'display_name': task.pretty_id,
            'status': task.status,
            'workers': list(task.workers),
            'worker_running': task.worker_running,
            'time_running': getattr(task, "time_running", None),
            'start_time': task.time,
            'last_updated': getattr(task, "updated", task.time),
            'params': task.params,
            'name': task.family,
            'priority': task.priority,
            'resources': task.resources,
            'tracking_url': getattr(task, "tracking_url", None),
            'status_message': task.status_message
        }
        if task.status == DISABLED:
            ret['re_enable_able'] = task.scheduler_disable_time is not None
        if include_deps:
            ret['deps'] = list(task.deps if deps is None else deps)
        return ret

    def graph(self, **kwargs):
        self.prune()
        serialized = {}
        seen = set()
        for task in self._state.get_active_tasks():
            serialized.update(self._traverse_graph(task.id, seen))
        return serialized

    def _filter_done(self, task_ids):
        for task_id in task_ids:
            task = self._state.get_task(task_id)
            if task is None or task.status != DONE:
                yield task_id

    def _traverse_graph(self, root_task_id, seen=None, dep_func=None, include_done=True):
        """ Returns the dependency graph rooted at task_id

        This does a breadth-first traversal to find the nodes closest to the