luigi scanner

scan.py

import inspect
import pkgutil
import sys
import bob.tasks
from datetime import date
from luigi.task import Task
from datetime import datetime


def explore_package(module_name):
    loader = pkgutil.get_loader(module_name)
    for sub_module in pkgutil.walk_packages([loader.filename]):
        _, sub_module_name, _ = sub_module
        qname = module_name + "." + sub_module_name
        yield qname
        explore_package(qname)
  
for m in reversed(list(explore_package("bob.tasks"))):
     clsmembers = inspect.getmembers(sys.modules[m], inspect.isclass)
     for cls in clsmembers:
         if issubclass(cls[1], Task):
             mem = dict(inspect.getmembers(cls[1]))
             params = {}
             if 'date' in mem:
                 params['date'] = date(2018, 1, 1)
             if 'look_back' in mem:
                 params['look_back'] = 1
             if 'date_hour' in mem:
                 params['date_hour'] = datetime(2018, 1, 1)
             t = cls[1](**params)
             o = t.output()
             print t.requires(), o.path if o != [] else []
            
            
            
            
            
            
         
  for m in reversed(list(explore_package("bob.tasks"))):
...     clsmembers = inspect.getmembers(sys.modules[m], inspect.isclass)
...     for cls in clsmembers:
...         if issubclass(cls[1], Task):
...             mem = set([i[0] for i in inspect.getmembers(cls[1])])
...             if 'date' in mem:
...                 print cls[1](date=today)
...             else:
...                 print cls[1]()