A luigi pipeline with multiple files

multi_file_task.py

import luigi

NUMBER_OF_FILES = 50


class WritePipelineTask(luigi.Task):

    def output(self):
        return luigi.LocalTarget("data/_DONE_WRITE_PIPELINE_TASK")

    def run(self):
        for i in range(NUMBER_OF_FILES):
            output_path = "data/output_one_{:d}.txt".format(i)
            with open(output_path, "w") as output_file:
                output_file.write("pipeline")

        open("data/_DONE_WRITE_PIPELINE_TASK", "a").close()


class AddMyTask(luigi.Task):

    def output(self):
        return luigi.LocalTarget("data/_DONE_ADD_MY_TASK")

    def requires(self):
        return WritePipelineTask()

    def run(self):
        for i in range(NUMBER_OF_FILES):
            input_path = "data/output_one_{:d}.txt".format(i)
            with open(input_path) as input_file:
                line = input_file.read()

            output_path = "data/output_two_{:d}.txt".format(i)
            with open(output_path, "w") as output_file:
                decorated_line = "My "+line
                output_file.write(decorated_line)

        open("data/_DONE_ADD_MY_TASK", "a").close()