luigi test

my_task.py

import datetime
import luigi

class TaskX(luigi.Task):
    x = luigi.IntParameter(default=777)

    def run(self):
        with self.output().open("w") as f:
            print >>f, self.x

    def output(self):
        return luigi.LocalTarget('/mapr/vgonzalez.solr/tmp/x')

class MyTask(luigi.Task):
    x = luigi.IntParameter()
    y = luigi.IntParameter(default=45)

    def requires(self):
        return TaskX()

    def run(self):
        with self.output().open("w") as f:
                print >>f, int(self.input().open("r").read()) + self.y

    def output(self):
        return luigi.LocalTarget('/mapr/vgonzalez.solr/tmp/output-%s' % datetime.datetime.now().isoformat())


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