luigi_task_template_advanced.py

import luigi
import os
#from WorkflowUtils import WorkflowUtils
import logging as log 

class ATask(luigi.Task):
    # Task parameters
    dataset_name = luigi.Parameter()
    upstream_task = luigi.Parameter()

    # Helper classes
    #utils = WorkflowUtils()

    def requires(self):
        return self.upstream_task

    def output(self):
        return luigi.LocalTarget(self.input().path + ".CHANGETHIS")
 
    def run(self):
        with gzip.open(self.input().path, "rb") as infile, gzip.open(self.output().path, "wb") as outfile:
            tsv_reader = csv.reader(infile, delimiter=" ")
            tsv_writer = csv.writer(outfile, delimiter=" ")
            for row in tsv_reader:
                outfile.write(row)