Demonstrates how one can select which parent task a luigi task should depend on, by specifying the parent task class name via a parameter.

select_parent_luigi_task_by_param.py

import luigi

class ATask(luigi.Task):

    def output(self):
        return luigi.LocalTarget("atask_output.txt")

    def run(self):
        with self.output().open("w") as outfile:
            outfile.write("Test")
        print "ATask was run"


class BTask(luigi.Task):

    def requires(self):
        return ATask()

    def output(self):
        return luigi.LocalTarget("btask_output.txt")

    def run(self):
        with self.output().open("w") as outfile:
            outfile.write("Test")
        print "BTask was run"


class CTask(luigi.Task):
    parent_class = luigi.Parameter()

    def get_class(self, class_name):
        return globals()[class_name]

    def requires(self):
        return self.get_class(self.parent_class)()

    def output(self):
        return luigi.LocalTarget("ctask_output.txt")

    def run(self):
        with self.output().open("w") as outfile:
            outfile.write("Test")
        print "CTask was run"


if __name__ == '__main__':
    luigi.run()

Now, CTask can either depend on ATask or BTask, by choosing one of the following commands:

python test.py --local-scheduler CTask --parent-class ATask

and:

python test.py --local-scheduler CTask --parent-class BTask

Randomly came across this gist googling something... Turn out in the last 7 years they've added a TaskParameter (:rofl:). For others who stumble across this an alternative implementation for CTask would be:

class CTask(luigi.Task):
    required_task = luigi.TaskParameter()
    def requires(self):
        # whatever args you need to pass through, keeping in mind:
        # `--BTask-some-param 'BTask Param Value'` can be used
        # for global assignment of BTask instances
        args = []
        return self.required_task(*args)