“未实现的运行时依赖”：创建动态依赖时出错

Question

当我试图从一个包含包含类名、文件名和“depend\u task”类的变量的列表中创建Luigi任务类时，我得到了上面的运行时错误。在

问题是类定义应该包括同时定义的“depend_task”类的定义。下面是我失败的尝试和错误消息。在

感谢您的帮助。在

class BDX_Query_0XX(SQLTask):
    acctDate = luigi.Parameter()
    ssisDate = luigi.Parameter() 
    runDesc = luigi.Parameter()

    def __init__(self, *args, **kwargs):
        super(BDX_Query_0XX, self).__init__(*args, **kwargs)

        # if self.trans_id is None:
        self.trans_id =  "00902_BDX_Query_0XX" + "__" + self.runDesc  # static.


    def run(self):
        YY = self.acctDate[:4]
        MM = self.acctDate[4:6]
        acctDate = self.acctDate
        ssisDate = self.ssisDate
        runDesc = self.runDesc
        bdx_sql = r'r:\\1.SQL\\BDX_SQL\\'
        cmdList = [
                ('BDX010',f'{bdx_sql}BDX_001_NI_DM 010.sql -o output010.txt',None),
                ('BDX020',f'{bdx_sql}BDX_001_NI_DM 020.sql -o output020.txt','BDX010'),
               ('BDX022a',f'{bdx_sql}BDX_022_P038_All_Final_CatAdj 010.sql -o output022a.txt','BDX020'),
            ('BDX022b',f'{bdx_sql}BDX_022_P038_All_Final_CatAdj 020.sql -o output022b.txt -v Year1={YY} MM={MM}','BDX022a'),
            ('BDX022c',f'{bdx_sql}BDX_022_P038_All_Final_CatAdj 030.sql -o output022c.txt -v Year={YY} Month={MM}', 'BDX022b'),
            ('BDX023',f'{bdx_sql}BDX_023_P031_MTD_All_Final_CatAdj.sql -o output023.txt ','BDX020'),
            ('BDX024',f'{bdx_sql}BDX_024_P031_ITD_All_Final_CatAdj.sql -o output024.txt','BDX020'),
            ('BDX025a',f'{bdx_sql}BDX_025_P038_All_Final_CatAdj 010.sql -o output025a.txt','BDX020'),
            ('BDX025b',f'{bdx_sql}BDX_025_P038_All_Final_CatAdj 020.sql -o output025b.txt -v Year={YY} Month={MM}','BDX025a'),
            ('BDX025c',f'{bdx_sql}BDX_025_P038_All_Final_CatAdj 030.sql -o output025c.txt -v YYMM={acctDate}','BDX025b')
            ]

        tasks = []
        for queryKey, queryCmd, dependQry in cmdList:
            class_name = queryKey
            klass = type(queryKey, (BDX_Task,),{}) 

            **#>>>> ERROR OCCURS IN THE FOLLOWING LINE >>>>>>>**
            tasks.append(klass(acctDate=self.acctDate, ssisDate =self.ssisDate, queryKey = queryKey, queryCmd = queryCmd, runDesc = self.runDesc, dependQry = dependQry)) # this addes Task class named after queryKey to dependency
        yield tasks

        self.get_target().touch()



class BDX_Task_Base(SQLTask):
    acctDate = luigi.Parameter()
    ssisDate = luigi.Parameter(default=None)
    queryKey = luigi.Parameter()
    queryCmd = luigi.Parameter()
    runDesc = luigi.Parameter()
    dependQry = luigi.Parameter()


    def __init__(self, *args, **kwargs):
        super(BDX_Task_Base, self).__init__(*args, **kwargs)

        # if self.trans_id is None:
        self.trans_id = f"00903_BDX_Query_{self.queryKey}__{self.runDesc}"


class BDX_Task(BDX_Task_Base):

    def requires(self):
        dependQry = self.dependQry
        if dependQry:
            if type(dependQry) is str:   # class name in string, then convert to SQLTask here; 
                depend_class= type(dependQry, (BDX_Task_Base,), {} )
                depend_task = [depend_class(acctDate = self.acctDate,ssisDate = self.ssisDate, queryKey = self.queryKey, queryCmd = self.queryCmd, runDesc = self.runDesc, dependQry = self.dependQry)]
            else:
                depend_task =[dependQry(acctDate= self.acctDate, ssisDate = self.ssisDate,queryKey = self.queryKey, queryCmd = self.queryCmd, runDesc = self.runDesc, dependQry = self.dependQry)]
        else:    #class itself
            depend_task = []

        return depend_task

    def run(self):
        print(subprocess.call(self.queryCmd, shell=True))
        self.get_target().touch()

全堆栈跟踪

^{pr2}$

Answer 1

BDX_Task是BDX_Task_Base的子类。在BDX_Task_Base的初始化中，有一个对super(BDX_Task)的调用。。。它是BDX_Task_Base，因此它再次运行init，它将再次调用super(BDX_Task)，依此类推。我假设您打算编写super(BDX_Task_Base)...，这不会导致无限递归。只要您使用的是最新版本的python，super().__init__(...)将执行与super(BDX_Task_Base, self).__init__(...)相同的操作。在