tac.py 文件源码

def test(self):
        # in this example, put your test_plugin/test.html template at airflow/plugins/templates/test_plugin/test.htm
        attributes = []
        data_table = []
        operator_data={}

        for classes in ALL:
            operator_data['name']=str(classes.__name__)
            operator_data['type']=str(inspect.getfile(classes).split("_")[-1].split(".")[0])
            operator_data['module']=str(inspect.getfile(classes).split("_")[-2])
            try:
                operator_data['args']=str(classes.arguments) if classes.arguments else "NA"
            except Exception:
                operator_data['args']= "NOT FOUND"

            operator_data['desc']=str(inspect.getdoc(classes))
            operator_data['loc']=str(inspect.getfile(classes))          
            data_table.append(copy.deepcopy(operator_data))       

        #data_table = json.dumps(data_table)
        return self.render("tac_plugin/tac.html",attributes=attributes,data_table=data_table)