luigi_test.py

import luigi
from glob import glob
import os
raw_zips = '/1002/*.gz'


class ExtractGZ(luigi.Task):
    document_id = luigi.Parameter()

    def output(self):
        return luigi.LocalTarget(self.document_id + '.txt')

    def run(self):
        print os.path.dirname(self.document_id)


class TestTask(luigi.WrapperTask):
    def requires(self):
        for fname in glob(raw_zips):
            #print fname
            yield ExtractGZ(fname)