How to output a zipfile in Luigi Task?

Question

It is ok to output() zip files like this:

def output(self):
    date_path = self.search['date_path']
    zip_fn = "data/%s/%s.zip" % (date_path, date_path)
    return luigi.LocalTarget(zip_fn)

But how to pas this zip in run() method?

class ZeroTask(luigi.Task):
    path_in = luigi.Parameter()
    textfiles = []
    path_to_zip = ''

    def requires(self):
        return []

    def run(self):

       # Get a bunch of text files
       # Do some manipulations with textfiles
       # Create a result.zip
       # self.path_to_zip = '~/Project/result.zip'

    def output(self):
        zip_fn = self.path_to_result.zip
        return luigi.LocalTarget(zip_fn)

What should I do in the run() method?

Answer 1

You should be able to use zipfile to build the file however you'd like.

class MyTask(luigi.Task):
    def output(self):
      date_path = self.search['date_path']
      zip_fn = "data/%s/%s.zip" % (date_path, date_path)
      return luigi.LocalTarget(zip_fn)

     def run(self):
        ztemp = tempfile.NamedTemporaryFile(mode='wb')
        z = zipfile.ZipFile(ztemp, 'w')

        #  build the zip file

        z.close()
        os.rename(ztemp.name, self.output().path)

From the docs on FileSystemTarget,