Luigi support 1

script1.py

#!/usr/bin/env python

import logging
import time

import luigi

logger = logging.getLogger(__name__)


class Init(luigi.Task):

    def output(self):
        return luigi.LocalTarget('./data/Inti/init')

    def run(self):
        time.sleep(1)
        with self.output().open('w') as f:
            f.write('ok')


class DailyUpdates(luigi.Task):
     id = luigi.Parameter()

     def requires(self):
        return Init()

     def output(self):
        return luigi.LocalTarget(path='./data/DailyUpdates/%s' % self.id)

     def run(self):
         time.sleep(1)
         with self.output().open('w') as f:
             f.write('ok')


class Learning1(luigi.Task):
     id = luigi.Parameter()

     def requires(self):
         return DailyUpdates(id=self.id)

     def output(self):
         return luigi.LocalTarget(path=f'./data/Learning1/{self.id}.csv')

     def run(self):
         time.sleep(1)
         with self.output().open('w') as f:
             f.write('ok')


class Predict(luigi.Task):
     id = luigi.Parameter()

     def requires(self):
         yield Learning1(id=self.id)

     def output(self):
         return luigi.LocalTarget(path=f'./data/Predict/{self.id}.csv')

     def run(self):
         time.sleep(1)
         with self.output().open('w') as f:
             f.write('ok')


class Main(luigi.Task):

     def requires(self):
        for id in range(3):
            yield Predict(id=id)

The command runs fine:

 export PYTHONPATH=/dir/of/script1.py
luigi --module script1 Main