#!/usr/bin/env python


# Finds all tasks and task outputs on the dependency paths from the given downstream task T
# up to the given source/upstream task S (optional). If the upstream task is not given,
# all upstream tasks on all dependancy paths of T will be returned.

# Terms:
# if  the execution of Task T depends on the output of task S on a dependancy graph,
#  T is called a downstream/sink task, S is called an upstream/source task.

# This is useful and practical way to find all upstream tasks of task T.
# For example suppose you have a daily computation that starts with a task named Daily.
# And suppose you have another task named Aggregate. Daily triggers a few tasks
# which eventually trigger Aggregate. Now, suppose you find a bug in Aggregate.
# You fixed the bug and now you want to rerun it, including all it's upstream deps.
#
# To do that you run:
#      bin/deps.py --module daily_module Aggregate --daily-param1 xxx --upstream-family Daily
#
# This will output all the tasks on the dependency path between Daily and Aggregate. In
# effect, this is how you find all upstream tasks for Aggregate. Now you can delete its
# output and run Aggregate again. Daily will eventually trigget Aggregate and all tasks on
# the way.
#
# The same code here might be used as a CLI tool as well as a python module.
# In python, invoke find_deps(task, upstream_name) to get a set of all task instances on the
# paths between task T and upstream task S. You can then use the task instances to delete their output or
# perform other computation based on that.
#
# Example:
#
# PYTHONPATH=$PYTHONPATH:/path/to/your/luigi/tasks bin/deps.py \
# --module my.tasks  MyDownstreamTask
# --downstream_task_param1 123456
# [--upstream-family MyUpstreamTask]
#


import luigi.interface
from luigi.contrib.ssh import RemoteTarget
from luigi.postgres import PostgresTarget
from luigi.s3 import S3Target
from luigi.target import FileSystemTarget
from luigi.task import flatten
from luigi import parameter
import sys
from luigi.cmdline_parser import CmdlineParser
import collections


def get_task_requires(task):
    return set(flatten(task.requires()))


def dfs_paths(start_task, goal_task_family, path=None):
    if path is None:
        path = [start_task]
    if start_task.task_family == goal_task_family or goal_task_family is None:
        for item in path:
            yield item
    for next in get_task_requires(start_task) - set(path):
        for t in dfs_paths(next, goal_task_family, path + [next]):
            yield t


class upstream(luigi.task.Config):
    '''
    Used to provide the parameter upstream-family
    '''
    family = parameter.Parameter(default=None)


def find_deps(task, upstream_task_family):
    '''
    Finds all dependencies that start with the given task and have a path
    to upstream_task_family

    Returns all deps on all paths between task and upstream
    '''
    return set([t for t in dfs_paths(task, upstream_task_family)])


def find_deps_cli():
    '''
    Finds all tasks on all paths from provided CLI task
    '''
    cmdline_args = sys.argv[1:]
    with CmdlineParser.global_instance(cmdline_args) as cp:
        return find_deps(cp.get_task_obj(), upstream().family)


def get_task_output_description(task_output):
    '''
    Returns a task's output as a string
    '''
    output_description = u"n/a"

    if isinstance(task_output, RemoteTarget):
        output_description = u"[SSH] {0}:{1}".format(task_output._fs.remote_context.host, task_output.path)
    elif isinstance(task_output, S3Target):
        output_description = u"[S3] {0}".format(task_output.path)
    elif isinstance(task_output, FileSystemTarget):
        output_description = u"[FileSystem] {0}".format(task_output.path)
    elif isinstance(task_output, PostgresTarget):
        output_description = u"[DB] {0}:{1}".format(task_output.host, task_output.table)
    else:
        output_description = u"to be determined"

    return output_description


def main():
    deps = find_deps_cli()
    for task in deps:
        task_output = task.output()

        if isinstance(task_output, dict):
            output_descriptions = [get_task_output_description(output) for label, output in task_output.iteritems()]
        elif isinstance(task_output, collections.Iterable):
            output_descriptions = [get_task_output_description(output) for output in task_output]
        else:
            output_descriptions = [get_task_output_description(task_output)]

        print u"   TASK: {0}".format(task)
        for desc in output_descriptions:
            print u"                       : {0}".format(desc)


if __name__ == '__main__':
    main()

@@ -1754,16 +1754,54 @@
 sk%5D%0A#%0A%0A%0A
+from __future__ import print_function%0A
 import l
@@ -3381,17 +3381,16 @@
 ption = 
-u
 %22n/a%22%0A%0A 
@@ -3463,17 +3463,16 @@
 ption = 
-u
 %22%5BSSH%5D %7B
@@ -3614,17 +3614,16 @@
 ption = 
-u
 %22%5BS3%5D %7B0
@@ -3731,17 +3731,16 @@
 ption = 
-u
 %22%5BFileSy
@@ -3854,17 +3854,16 @@
 ption = 
-u
 %22%5BDB%5D %7B0
@@ -3952,17 +3952,16 @@
 ption = 
-u
 %22to be d
@@ -4520,26 +4520,25 @@
        print
- u
+(
 %22   TASK: %7B0
@@ -4552,16 +4552,17 @@
 at(task)
+)
 %0A       
@@ -4612,18 +4612,17 @@
    print
- u
+(
 %22       
@@ -4656,16 +4656,17 @@
 at(desc)
+)
 %0A%0A%0Aif __

Convert to Py3 print function and remove unicode strings