Implementation:Spotify Luigi Deps Tree Tool

Overview

The deps_tree tool in luigi/tools/deps_tree.py is a CLI utility and Python module that prints a visual ASCII tree of a task's dependency graph. It shows each task's name, significant parameters, and completion status (PENDING or COMPLETE) in a hierarchical tree format with ANSI color coding. This tool parses commands exactly like the Luigi task runner but instead of executing the task, it displays the execution plan.

Source Location

Property	Value
Source File	luigi/tools/deps_tree.py
Lines of Code	74
Module	luigi.tools.deps_tree
Domain	Debugging, Visualization
CLI Entry Point	luigi-deps-tree

Import Statement

from luigi.tools.deps_tree import print_tree

Classes

bcolors

bcolors

A simple class containing ANSI escape code constants for terminal color output.

Constant	Value	Usage
OKBLUE	'\033[94m'	Applied to PENDING task status text.
OKGREEN	'\033[92m'	Applied to COMPLETE task status text.
ENDC	'\033[0m'	Resets terminal color back to default.

Functions

print_tree

print_tree(task, indent='', last=True)

Parameter	Type	Default	Description
task	luigi.Task	(required)	The root task to visualize.
indent	str		Current indentation prefix string (used internally for recursion).
last	bool	True	Whether this is the last child in its parent's children list (controls tree branch characters).

Returns a string containing the ASCII tree representation of the task and all its recursive dependencies. The function works recursively:

1. Checks task.complete() to determine status (suppresses warnings for tasks without outputs).
2. Formats the task as [TaskName-{params} (STATUS)] where:
   • TaskName is task.__class__.__name__
   • {params} are the significant parameters from task.to_str_params(only_significant=True)
   • STATUS is either COMPLETE (green) or PENDING (blue)
3. Uses tree branch characters:
   • Last child in a list uses the corner branch
   • Non-last children use the T-branch |---
4. Recursively processes all children from flatten(task.requires()).

main

main()

The CLI entry point function. Parses sys.argv[1:] using CmdlineParser, constructs the task object, and prints the tree.

CLI Usage

# Display the dependency tree for a task
luigi-deps-tree --module my_module MyTask --param value

# Example with a complex dependency graph
luigi-deps-tree --module foo_complex examples.Foo

Output Format

The output is a color-coded ASCII tree showing task status:

└─--[Foo-{} (PENDING)]
    |---[Bar-{'num': '0'} (PENDING)]
    |   |---[Bar-{'num': '4'} (PENDING)]
    |   └─--[Bar-{'num': '5'} (PENDING)]
    |---[Bar-{'num': '1'} (PENDING)]
    └─--[Bar-{'num': '2'} (PENDING)]
        └─--[Bar-{'num': '6'} (PENDING)]
            |---[Bar-{'num': '7'} (PENDING)]
            |   |---[Bar-{'num': '9'} (COMPLETE)]
            |   └─--[Bar-{'num': '10'} (PENDING)]
            |       └─--[Bar-{'num': '11'} (COMPLETE)]
            └─--[Bar-{'num': '8'} (PENDING)]
                └─--[Bar-{'num': '12'} (COMPLETE)]

In a terminal:

• COMPLETE tasks are displayed in green (\033[92m)
• PENDING tasks are displayed in blue (\033[94m)

Tree Characters

Character Sequence	Meaning
---	A non-last child branch (sibling follows)
The corner branch	The last child in a sibling group
|	Vertical continuation line for non-last children
(4 spaces)	Indentation for children of the last child

Programmatic Usage

from luigi.tools.deps_tree import print_tree
import luigi

class TaskA(luigi.Task):
    def requires(self):
        return [TaskB(), TaskC()]
    def complete(self):
        return False

class TaskB(luigi.Task):
    def complete(self):
        return True

class TaskC(luigi.Task):
    def complete(self):
        return False

# Generate the tree string
tree_output = print_tree(TaskA())
print(tree_output)

Dependencies

• Luigi core: luigi.task.flatten, luigi.cmdline_parser.CmdlineParser
• Python standard library: sys, warnings

Related Principles

• Spotify_Luigi_Dependency_Analysis

See Also

• Spotify_Luigi_Deps_Tool - Finds all tasks on dependency paths between two tasks
• Spotify_Luigi_LuigiGrep_Tool - Search for tasks via the scheduler API