I don’t think you need the init also. Just let the Flow call the new

run

method. And then last in order to use it in the Flow,

ETL()

is just calling the init. You want to change

ETL()

to

ETL()()

, and the second

()

will call the run method.

I think this should work:

class Extract(Task):

    def run(self):
        read_conn = db.read_conn
        query ="""SELECT b.oproduct_id, p.oproduct_id,p.oproduct_parent_id,b.obundle_parent_id
        from hq.oproducts p 
        JOIN hq.obundles b ON b.oproduct_id = p.oproduct_id
        WHERE b.oproduct_id = 5801"""
        df = pd.read_sql(query,read_conn)
        return df

class Load(Task):
    def __init__(self, df):
        self.df = df
        super().__init__()

    def run(self):
        self.df.to_csv(r"C:\Users\<http://cho.ng|cho.ng>\test\df.csv",index=False)

with Flow('flow_3',executor=LocalDaskExecutor(), run_config=LocalRun()) as flow:
    df = Extract()()
    Load(df)()

flow.register(project_name="tester")

the load class failed

Let me do a quick test on my end and get back to you

Change to this:

class Load(Task):

    def run(self, df):
        df.to_csv(r"C:\Users\<http://cho.ng|cho.ng>\test\df.csv",index=False)
        
with Flow('flow_3',executor=LocalDaskExecutor(), run_config=LocalRun()) as flow:
    df = Extract()()
    Load()(df)

This will move the df to the runtime by passing it to the

run

method.

my question is, do I have to do it this way that limits to using only the run method in a class?

oh

This might work for you:

class ETL(Task):

    def extract(self):
        read_conn = db.read_conn
        query ="""SELECT b.oproduct_id, p.oproduct_id,p.oproduct_parent_id,b.obundle_parent_id
                from hq.oproducts p 
                JOIN hq.obundles b ON b.oproduct_id = p.oproduct_id
                WHERE b.oproduct_id = 5801"""
        df = pd.read_sql(query,read_conn)
        return df

    def load(self):
        self.df.to_csv(r"C:\Users\<http://cho.ng|cho.ng>\test\df.csv",index=False)

    def run(self):
        df = self.extract()
        self.load()

with Flow('flow_3',executor=LocalDaskExecutor(), run_config=LocalRun()) as flow:
    ETL()()